Readme (Readme.md)

System Design Interview - An Insider's Guide (Vol 1 and 2)

These notes are based on the System Design Interview books - Vol 1 and Vol 2 2nd Ed

Check the notes here: https://pagefy.io/system-design/system-design-interview-by-alex-xu

Note: These notes are a work in progress.

* Chapter 1 - Scale From Zero To Millions Of Users
* Chapter 2 - Back-of-the-envelope Estimation
* Chapter 3 - A Framework For System Design Interviews
* Chapter 4 - Design A Rate Limiter
* Chapter 5 - Design Consistent Hashing
* Chapter 6 - Design A Key-Value Store
* Chapter 7 - Design A Unique ID Generator In Distributed Systems
* Chapter 8 - Design A URL Shortener
* Chapter 9 - Design A Web Crawler
* Chapter 10 - Design A Notification System
* Chapter 11 - Design A News Feed System
* Chapter 12 - Design A Chat System
* Chapter 13 - Design A Search Autocomplete System
* Chapter 14 - Design YouTube
* Chapter 15 - Design Google Drive
* Chapter 16 - Proximity Service
* Chapter 17 - Nearby Friends
* Chapter 18 - Design Google Maps
* Chapter 19 - Distributed Message Queue
* Chapter 20 - Metrics Monitoring and Alerting System
* Chapter 21 - Ad Click Event Aggregation
* Chapter 22 - Hotel Reservation System
* Chapter 23 - Distributed Email Service
* Chapter 24 - S3-like Object Storage
* Chapter 25 - Real-time Gaming Leaderboard
* Chapter 26 - Payment System
* Chapter 27 - Digital Wallet
* Chapter 28 - Stock Exchange

Additonal Resources

---

01. Scaling/Readme (01. Scaling/Readme.md)

Chapter 1: Scale from Zero to Millions of Users

Introduction

Scaling a system to support millions of users is a complex, iterative journey requiring refinement and optimization. This chapter outlines how to begin with a single server setup and scale the architecture step by step to handle millions of users.

---

Section 1: Single Server Setup

Initially, all components (web app, database, cache) run on a single server.

Request Flow

1. Users access the application via domain names (e.g., api.mysite.com), resolved to IP addresses using DNS.
2. IP address of the web-server is returned to the browser or mobile app.
3. HTTP requests are sent to the web server, which returns HTML or JSON responses.

Traffic Sources

1. Web Applications: Use server-side languages (e.g., Python, Java) for business logic and client-side languages (e.g., JavaScript, HTML) for presentation.
2. Mobile Applications: Communicate with the web server using HTTP and JSON for lightweight data exchange.

---

Section 2: Database Separation

As the user base grows, the database is moved to a dedicated server to allow independent scaling of web and database tiers.

Database Choices

1. Relational Databases (SQL): Structured data stored in tables. Examples: MySQL, PostgreSQL.
2. Non-Relational Databases (NoSQL): Suitable for unstructured data or low-latency requirements. Categories include:
- Key-Value Stores
- Graph Databases
- Column Stores
- Document Stores

- Non-relational databases might be the right choice if:
- application requires super-low latency.
- data is unstructured, or there is no relational data.
- only need to serialize and deserialize data (JSON, XML, YAML, etc.).
- need to store a massive amount of data.

---

Section 3: Vertical vs Horizontal Scaling

Vertical Scaling

- Adds more resources (CPU, RAM) to existing servers.
- Limited by hardware constraints and lacks redundancy.

Horizontal Scaling

- Adds more servers to the pool, making it more suitable for large-scale systems.
- A load balancer is used to handle the request routing between the servers.
---

Section 4: Load Balancer

A load balancer distributes traffic among multiple servers. Benefits include:
1. Redundancy: If a server goes offline, traffic is rerouted.
- If server 1 goes offline, all the traffic will be routed to server 2.
2. Scalability: Easily add servers to handle traffic spikes.
- If the website traffic grows rapidly, subsequent servers can be added to handle the additional traffic.

---

Section 5: Database Replication

Master-Slave Model

- Master Database: Handles write operations.
- All the data-modifying commands like insert, delete, or update must be sent to the master database.
- Slave Databases: Handle read operations, improving performance and reliability.
- Since the ratio of reads to writes is higher in most applications; thus, the number of slave
databases in a system is usually larger than the number of master databases.

Benefits

1. Improved performance through parallel read operations.
2. High availability and data reliability through redundancy.

Failure Handling

- If only one slave database is available and it goes offline, read operations will be directed
to the master database temporarily.
- In case multiple slave databases are available, read operations are
redirected to other healthy slave databases and a new server will replace the old one.
- If the master database goes offline, a slave database will be promoted to be the new
master.
- In production system the chosen slave database might not be up to date, hence data needs to be updated by running data
recovery scripts (methods like multi-masters and circular replication could help).

---

Section 6: Caching

A cache stores frequently accessed data in memory to reduce database load. The cache tier is a temporary data store layer, much faster than the database.

Caching considerations

1. Use case: Consider using cache when data is read frequently but modified infrequently.
2. Expiration Policies: Once cached data is expired, it is removed from the cache. When there is no expiration policy, cached
data will be stored in the memory permanently.
3. Consistency: This means keeping the data store and the cache in sync. Inconsistency
can happen because data-modifying operations on the data store and cache are not in a single transaction.
4. Mitigating failures: A single cache server represents a potential single point of failure, multiple
cache servers across different data centers are recommended to avoid SPOF.
5. Eviction Policies:: Once the cache is full, items need to be evicted to free up memory. LRU is the most popular cache eviction policy.

---

Section 7: Content Delivery Network (CDN)

A CDN improves load times by caching static content (images, CSS, JavaScript) on geographically distributed servers.

Workflow

1. User requests content from the nearest CDN server.
2. If unavailable, content is fetched from the origin server and cached.

CDN considerations

1. Cost: CDNs are run by third-party providers which charge for data transfers in and out of the CDN.
2. Cache Expiry: The cache expiry time should neither be too long nor too short.
3. CDN fallback: If there is a temporary CDN outage, clients should be able to detect the problem
and request resources from the origin.
4. Invalidating files: If files are updated the cache should be invalidated to point to the updated files.

---

Section 8: Stateless Web Tier

By moving session data to a shared datastore, web servers become stateless. This allows:
1. Easier horizontal scaling.
2. Auto-scaling based on traffic.

---

Section 9: Multi-Data Center Setup

Deploying across multiple data centers improves availability and reduces latency. Strategies include:

1. GeoDNS Routing: Direct users to the nearest data center.
2. Data Replication: Synchronize data across centers to prevent inconsistencies.

Key considerations

- Traffic redirection: Effective tools are needed to direct traffic to the correct data center.
- Data synchronization: A common strategy is to replicate data across multiple data centers.
- Test and deployment: Automated deployment tools are vital to keep services consistent through all the data centers.

---

Section 10: Message Queue

A message queue is a durable component, stored in memory, that supports asynchronous
communication. It serves as a buffer and distributes asynchronous requests.

- Input services, called producers/publishers, create messages, and publish them to a message queue.
- Other services called consumers/subscribers, connect to the queue, and perform actions defined by the messages.

---

Section 11: Logging, Metrics, and Automation

Importance

1. Logging: Tracks errors and system health.
2. Metrics: Provides insights into performance and user activity.
3. Automation: Streamlines testing, deployment, and scaling.

---

Section 12: Database Scaling

Vertical Scaling

- Adds hardware resources but has physical and cost limitations.
- Has multiple drawbacks:
- Greater risk of single point of failures.
- Overall cost of vertical scaling is high

Horizontal Scaling (Sharding)

- Divides data across multiple shards using keys (e.g., user_id).
- Sharding separates large databases into smaller, more easily managed parts called shards.
- Each shard shares the same schema, though the actual data on each shard is unique to the shard.
- Sharding key is critical when implementing a sharding strategy. When choosing a sharding key it is important to choose a key that can evenly distribute data.

#### Challenges
1. Resharding data: Resharding data is needed when:
- Single shard could no longer hold more data due to rapid growth.
- Certain shards might experience shard exhaustion faster than others due to uneven data distribution.
- Consistent Hashing is used to overcome these problems

2. Celebrity problem: Excessive access to a specific shard could cause server overload.
- To solve this problem, we may need to allocate a shard for each celebrity.

3. Join and de-normalization: Once a database has been sharded across multiple servers, it is hard to perform join operations across database shards.
- A common workaround is to de-normalize the database so that queries can be performed in a single table.

---

Conclusion

Key Takeaways

1. Keep the web tier stateless.
2. Build redundancy at every tier.
3. Use caching and CDNs to optimize performance.
4. Scale the data tier with sharding.
5. Decouple components for flexibility.

This chapter provides a solid foundation for building scalable systems that can handle millions of users.

---

02. Back Of The Envelope Estimation/Readme (02. Back Of the Envelope Estimation/Readme.md)

Chapter 2: Back-of-the-Envelope Estimation

Introduction

Back-of-the-envelope estimation is a crucial skill in system design interviews. It involves making quick, rough calculations to assess system capacity or performance. According to Jeff Dean, Google Senior Fellow, these estimates help evaluate whether designs meet requirements through thought experiments and common performance benchmarks.

This chapter covers key concepts, methodologies, and examples to build proficiency in scalability and estimation.

---

Section 1: Key Concepts

Power of Two

Understanding data volume in terms of powers of two is fundamental:

This knowledge helps in performing accurate storage and bandwidth calculations.

---

Latency Numbers Every Programmer Should Know

Latency numbers represent the time taken for various operations in computing systems. These provide insights into relative performance:

| Operation | Latency (2020) |
|--------------------------|----------------|
| L1 Cache Access | 0.5 ns |
| L2 Cache Access | 7 ns |
| Main Memory Access | 100 ns |
| SSD Random Read | 150 µs |
| HDD Random Seek | 10 ms |
| Round-Trip in Data Center| 500 µs |
| Inter-Region Data Center | 150 ms |

Key Insights:
- Memory is fast, disk is slow.
- Avoid disk seeks whenever possible.
- Compress data before transmitting over the internet to save bandwidth.

---

Availability Numbers

High availability (HA) ensures minimal downtime. Availability is expressed in nines:
- 99% (Two Nines): ~3.65 days/year of downtime
- 99.9% (Three Nines): ~8.8 hours/year of downtime
- 99.99% (Four Nines): ~52 minutes/year of downtime
- 99.999% (Five Nines): ~5.3 minutes/year of downtime
- 99.9999% (Six Nines): ~31.56 seconds/year of downtime

Cloud providers like Amazon, Google, and Microsoft aim for SLAs (Service Level Agreements) of 99.9% or higher.

---

Section 2: Example Estimation - Twitter QPS and Storage Requirements

Assumptions

- 300 million monthly active users (MAU).
- 50% daily active users (DAU).
- Average tweets/user/day: 2.
- 10% of tweets contain media.
- Data retention: 5 years.

Estimations

1. Query Per Second (QPS):
- DAU = $ 300M x 50\% = 150M $
- Tweets QPS = $ 150M x 2 tweets / 24 hour / 3600 seconds = ~3500 )
- Peak QPS = \( 2 x 3500 = ~7000 $

2. Media Storage:
- Tweet Size Components:
- tweet_id: 64 bytes
- text: 140 bytes
- media: 1 MB
- Daily Media Storage: $ 150M x 2 x 10\% x 1MB = 30TB per day $
- 5-Year Storage: $ 30TB x 365 x 5 = ~55PB $

---

Section 3: Tips for Effective Estimation

1. Rounding and Approximation

Precision is not critical; focus on the process. Simplify complex calculations using round numbers. For example:
- $ 99987 / 9.1 $ can be approximated as $ 100,000 / 10 = 10,000 $.

2. Write Down Assumptions

Document assumptions clearly for future reference.

3. Label Units

Avoid ambiguity by labeling units (e.g., 5 MB instead of 5).

4. Common Estimation Scenarios

- QPS (Queries Per Second): Measure traffic intensity.
- Peak QPS: Account for traffic spikes.
- Storage Requirements: Estimate total data needs.
- Cache Requirements: Evaluate memory requirements for caching.
- Number of Servers: Calculate hardware needs based on workload.

---

03. System Design Framework/Readme (03. System Design Framework/Readme.md)

Chapter 3: A Framework for System Design Interviews

Introduction

System design interviews are a key part of the hiring process, simulating real-life problem-solving scenarios. These interviews evaluate not just technical skills but also collaboration, communication, and the ability to handle ambiguous requirements.

This chapter introduces a 4-step framework for navigating system design interviews effectively.

---

Step 1: Understand the Problem and Establish Design Scope

Key Objectives

- Clarify requirements and assumptions.
- Avoid jumping into solutions prematurely.
- Showcase critical thinking by asking good questions.

Approach

- Ask Clarifying Questions:
- What are the most important features?
- What scale does the system need to handle?
- Are we building for web, mobile, or both?
- Are there existing technologies or constraints?

- Document Assumptions: Write assumptions on a whiteboard or paper for reference.

Example

Problem: Design a news feed system.
Questions:
- Is it a mobile app, web app, or both?
- How many friends can a user have?
- Should the feed include images and videos?
- Is the feed sorted by reverse chronological order?

---

Step 2: Propose High-Level Design and Get Buy-In

Key Objectives

- Develop a high-level architecture.
- Collaborate with the interviewer to refine the design.

Approach

- Draft a Blueprint:
- Use box diagrams for key components (e.g., clients, APIs, databases, caches, CDNs).
- Treat the interviewer as a teammate to refine the design.

- Perform Back-of-the-Envelope Calculations:
- Ensure the design can handle the scale constraints.

- Walk Through Use Cases: Identify edge cases and validate design assumptions.

Example

For a news feed system, divide the design into:
1. Feed Publishing Flow: Writing posts into databases and populating friends' feeds.
2. Feed Retrieval Flow: Aggregating and displaying friends' posts in reverse chronological order.

---

Step 3: Design Deep Dive

Key Objectives

- Dive into critical components.
- Showcase depth of understanding and adaptability.

Approach

- Prioritize Key Components: Focus on areas most relevant to the problem.
- Discuss Bottlenecks: Identify potential performance issues and propose solutions.
- Balance Detail: Avoid over-engineering or unnecessary deep dives.

Example Topics

- URL Shortener: Focus on hash function design.
- Chat System: Explore latency reduction and online/offline status handling.
- News Feed System: Examine feed publishing and retrieval processes.

---

Step 4: Wrap-Up

Key Objectives

- Highlight areas for improvement.
- Recap the design and discuss follow-ups.

Approach

- Identify Bottlenecks: Discuss potential limitations and scaling strategies.
- Summarize Design: Recap major design decisions and trade-offs.
- Propose Enhancements:
- How to scale from 1 million to 10 million users.
- Error handling for server failures or network issues.

---

Best Practices

Dos

- Ask Questions: Clarify ambiguities before diving into solutions.
- Communicate: Share your thought process with the interviewer.
- Iterate with the Interviewer: Treat them as a collaborator.
- Show Flexibility: Suggest alternative approaches and refine your design.
- Focus on Critical Components: Prioritize key parts of the system.

Don’ts

- Avoid Premature Solutions: Don’t design before understanding the requirements.
- Don’t Go Silent: Communicate regularly during the process.
- Avoid Over-Engineering: Focus on practical, scalable solutions.

---

Time Management

Suggested Time Allocation (for 45-Minute Interviews):

1. Understand Problem and Scope: 3–10 minutes
2. High-Level Design and Buy-In: 10–15 minutes
3. Deep Dive: 10–25 minutes
4. Wrap-Up: 3–5 minutes

---

04. Rate Limiter/Readme (04. Rate Limiter/Readme.md)

Chapter 4: Design a Rate Limiter

Introduction

This chapter explores the design and implementation of a rate limiter—a system component used to control traffic rates sent by clients or services. Rate limiters are crucial for preventing abuse, reducing costs, and ensuring the stability of server resources. Examples of their use include limiting posts, account creations, and reward claims.

Benefits of Rate Limiting

- Preventing DoS Attacks: Blocking excess calls to avoid resource starvation.
- Cost Reduction: Limiting unnecessary requests to reduce server expenses.
- Preventing Overloads: Filtering out excessive requests to stabilize server performance.

Step 1: Understanding the Problem

Key Features

- Server-side API rate limiter.
- Support for multiple throttle rules.
- Handle large-scale systems in distributed environments.
- Option for a standalone service or application-level code.
- Inform users when throttled.

Requirements

- Accurate request throttling.
- Minimal latency.
- Low memory usage.
- Distributed capability.
- Clear exception handling.
- High fault tolerance.

Step 2: High-Level Design

Placement Options

1. Client-Side Implementation: Unreliable due to potential misuse.
2. Server-Side Implementation: Preferred for control and reliability.
3. Middleware (API Gateway): A flexible option for integrated rate limiting.

Guidelines for Placement

- Evaluate current tech stack and choose efficient options.
- Select appropriate algorithms based on business needs.
- Use an API gateway if microservices are employed.
- Opt for commercial solutions if resources are limited.

Step 3: Rate Limiting Algorithms

1. Token Bucket

- Description: Tokens are added to a bucket at a fixed rate; each request consumes a token.
- Parameters: Bucket size and refill rate.
- Pros: Easy to implement, memory-efficient, supports traffic bursts.
- Cons: Requires careful parameter tuning.

2. Leaking Bucket

- Description: Processes requests at a fixed rate using a FIFO queue.
- Pros: Memory-efficient, stable outflow rate.
- Cons: Traffic bursts may delay recent requests.

Example: https://github.com/uber-go/ratelimit

3. Fixed Window Counter

- Description: Divides time into fixed intervals and uses counters to limit requests.
- Pros: Simple, efficient for specific use cases.
- Cons: Traffic spikes at window edges can exceed limits.

- Sudden burst of traffic at the edges of time windows
could cause more requests than allowed quota to go through.

4. Sliding Window Log

- Description: Tracks timestamps to allow a rolling time window.
- Pros: Accurate rate limiting.
- Cons: High memory consumption.

5. Sliding Window Counter

- Description: Combines fixed window and sliding log methods for smoothing spikes.
- Pros: Memory-efficient, handles traffic bursts.
- Cons: Approximation may not be perfectly strict.

High-Level Architecture

- Data Storage: Use in-memory caching (e.g., Redis) for fast counter operations.
- Steps:
1. Client sends request to middleware.
2. Middleware checks counters in Redis.
3. Request is processed or rejected based on limits.

Advanced Considerations

Distributed Environments

- Challenges: Race conditions, synchronization issues.
- Solutions: Use locks, Lua scripts, or sorted sets in Redis. Employ centralized data stores for synchronization.

Performance Optimizations

- Multi-data center setups for reduced latency.
- Eventual consistency models for synchronization.

Monitoring

- Regular analytics to ensure algorithm effectiveness and adjust rules as needed.

---

05. Consistent Hashing/Readme (05. Consistent Hashing/Readme.md)

Chapter 5: Design Consistent Hashing

Introduction

This chapter explores consistent hashing, a technique essential for achieving horizontal scaling by efficiently distributing requests and data across servers. It minimizes data redistribution when servers are added or removed and ensures an even distribution of data to mitigate issues like server hotspots.

The Rehashing Problem

Explanation

In traditional hashing methods, such as serverIndex = hash(key) % N, data redistribution becomes problematic when the number of servers changes. For example:
- Removing a server causes most keys to be reassigned, leading to cache misses.
- Adding a server results in unnecessary key redistributions.

- This approach works well when the size of the server pool is fixed. However, problems arise when new servers are added, or existing servers are removed.

Key Issue

Redistribution of most keys when server count changes causes inefficiency and overload.

Consistent Hashing

Definition

Consistent hashing ensures that only a fraction of keys are remapped when servers are added or removed. This minimizes disruptions and enhances scalability.

Key Concepts

1. Hash Space and Ring: The hash space forms a continuous ring, with hash values distributed from 0 to 2^160-1 (e.g., using hash function like SHA-1). By connecting both ends we get a ring.
<p align="center">
<img src="./images/hash-ring.png" alt="Hash Ring" width="450">
</p>

- Using the same hash function f, we map servers based on server IP or name onto the ring.

1. Server Lookup
- A key's server is determined by traversing clockwise on the ring until a server is found.

2. Adding and Removing Servers
- Adding a server redistributes only nearby keys. Only a fraction of keys are redistributed to the new server.

<p align="center">
<img src="./images/adding-server.png" alt="Adding Server" width="450">
</p>

- Removing a server affects only the keys in its range. Only keys from the removed server are reassigned to the next server clockwise.

Challenges and Solutions

Two Issues in Basic Approach

1. Uneven Partition Sizes: Servers may have unequal data partitions.
2. Non-uniform Key Distribution: Some servers may receive significantly more keys than others.

Solution: Virtual Nodes

- Each server is represented by multiple virtual nodes on the ring uniformly distrubuted on the ring.
- Virtual nodes improve key distribution and balance load. As the number of virtual nodes increases, the distribution of keys becomes more balanced. This is because the standard deviation gets smaller with more virtual nodes, leading to balanced data distribution.

<p align="center">
<img src="./images/virtual-nodes.png" alt="Virtual Nodes" width="450">
</p>

Affected Keys

When servers are added or removed:
- Added Server: Affected keys are those between the new server and its predecessor.
In the following example server 4 is added onto the ring. The affected range starts from s4 (newly
added node) and moves anticlockwise around the ring until a server is found (s3). Thus, keys
located between s3 and s4 need to be redistributed to s4.

- Removed Server: Affected keys are those between the removed server and its predecessor. In the following example when a server (s1) is removed, the affected range starts from s1
(removed node) and moves anticlockwise around the ring until a server is found (s0). Thus, keys located between s0 and s1 must be redistributed to s2.

<p align="center">
<img src="./images/server-removed.png" alt="Server Removed" width="450">
</p>

Benefits of Consistent Hashing

- Minimized Redistribution: Only a fraction of keys are reassigned.
- Scalability: Enables horizontal scaling.
- Mitigates Hotspots: Balances data distribution to avoid server overload.

Real-World Applications

- Amazon Dynamo DB
- Apache Cassandra
- Discord
- Akamai CDN
- Maglev Load Balancer

---

06. Key Value Store/Readme (06. Key-Value Store/Readme.md)

Chapter 6: Design a Key-Value Store

Introduction

A key-value store is a type of non-relational database where data is stored as key-value pairs. Each key is unique, and values are accessed using these keys. This chapter details how to design a scalable, high-availability distributed key-value store that supports operations like:
- put(key, value) for inserting data.
- get(key) for retrieving data.

Characteristics of the Design

- Small key-value pairs (<10 KB).
- Supports big data with high availability and scalability.
- Automatic scaling and tunable consistency.
- Low latency.

---

Single Server Key-Value Store

Implementation

- Use a hash table to store key-value pairs in memory.
- Optimizations:
- Data compression.
- Storing less frequently accessed data on disk.

Limitation

A single server's memory is limited, requiring a distributed approach for scalability.

---

Distributed Key-Value Store

A distributed key-value store partitions data across multiple servers and must address trade-offs outlined by the CAP theorem.

CAP Theorem

1. Consistency: All clients see the same data simultaneously.
2. Availability: The system responds to every request, even if some nodes are down.
3. Partition Tolerance: The system continues to operate despite network partitions.

Trade-off: According to CAP theorem only two of the three guarantees can be achieved.

#### System Types:
- CP Systems: Consistency and partition tolerance while sacrificing availability (e.g., banking systems).
- AP Systems: Availability and partition tolerance while sacrificing consistency (e.g., eventual consistency).
- CA Systems: Consistency and Availability while sacrificing partition tolerance.

Since network failure is unavoidable, a distributed system must tolerate network partition. Thus, a CA system cannot exist in real-world applications.

In a distributed system, partitions are inevitable. When a partition occurs, we must choose between consistency and availability. For example, if node n3 goes down,
any data written to nodes n1 or n2 cannot be propagated to n3. Conversely, if data is written to n3 but not yet propagated to n1 and n2, nodes n1 and n2 will have stale data.

<p align="center">
<img src="./images/server-down.png" alt="Server down" width="400">
</p>

- If we choose CP system, we must block all write operations to n1 and n2 to avoid data inconsistency.
- If we choose AP system, the system keeps accepting reads, even though it might return stale data.
For writes, n1 and n2 keep accepting writes,
and data will be synced to n3 when the network partition is resolved.

---

System Components

1. Data Partitioning

- Technique: Consistent Hashing is used to distribute data across multiple servers evenly.
- Advantages:
- Automatic scaling with server addition/removal.
- Heterogeneity through virtual nodes. The number of virtual nodes for a server is proportional to the server capacity.

2. Data Replication

- Replicate data across N servers for high availability.
- The N servers are chosen by walking clockwise from the server position and choose the first N servers on the ring to store data copies.Place replicas in distinct data centers to improve reliability in case of virtual nodes.

3. Consistency

Since data is replicated at multiple nodes, it must be synchronized across replicas.
- Quorum Consensus:
- N: Total replicas.
- W: Write quorum size. For a write to be considered successful, write must be acknowledged from W replicas.
- R: Read quorum size. For a read to be considered as successful, read must wait for responses from at least R replicas.
- Rule: W + R > N ensures strong consistency.
- The configuration of W, R and N is a typical tradeoff between latency and consistency.

<p align="center">
<img src="./images/quorum-consensus.png" alt="Quorum consensus" width="400">
</p>

- If R = 1 and W = N, the system is optimized for a fast read.
- If W = 1 and R = N, the system is optimized for fast write.
- If W + R > N, strong consistency is guaranteed (Usually N = 3, W = R = 2).
- If W + R <= N, strong consistency is not guaranteed.

- Models:
- Strong Consistency: A read operation returns a value corresponding to the result of the most updated write data item.
- Weak Consistency: Subsequent read operations may not see the most updated value.
- Eventual Consistency: Given enough time, all updates are propagated, and all replicas are consisten

4. Inconsistency Resolution

Replication gives high availability but causes inconsistencies among replicas. Versioning and
vector locks are used to solve inconsistency problems.
- Versioning:
- Use vector clocks to track data versions and resolve conflicts.
- Versioning means treating each data modification as a new immutable version of data.
<div>
<img src="./images/consistent-server.png" alt="Consisten hashing" width="400">
<img src="./images/inconsistent-server.png" alt="Inconsistent server" height="230">
</div>

- Server 1 changes the name , and server 2 also changes the name. These two changes are performed simultaneously. Now, we have conflicting values, called versions v1 and v2.

- Vector Clock
1. Setup: A vector clock is a [server, version] pair associated with a data item. It can be used to check
if one version precedes, succeeds, or in conflict with others.
- Assume a vector clock represented by D([S1, v1], [S2, v2], …, [Sn, vn]), If data item D is written to server
Si, the system must perform one of the following tasks.
- Where: D is the data item.Si is the server identifier.vi is the version counter for the data at server Si.

2. Updating the Vector Clock: When a data item is modified at a server:
- If the server exists in the vector clock, its version counter is incremented.
- Otherwise, a new entry is added to the vector clock.

3. Conflict Detection:
- No Conflict: A version X is an ancestor of version Y if all counters in X are less than or equal to those in Y.
- Conflict Exists: Two versions are siblings if there is at least one counter in Y that is less than its counterpart in X.

4. Conflict Resolution: When conflicts are detected (sibling versions), the system relies on application-specific logic or client intervention to reconcile the data.

- Challenges:
- Increased complexity for clients.
- Vector clock size may grow with many updates, requiring trimming strategies to limit its size.

5. Handling Failures

#### a. Failure Detection
It is insufficient to believe that a server is down because another server says so.Usually, it requires at least two independent sources of information to mark a server down.
- Gossip Protocol:
<div style="margin-left:3rem">
<img src="./images/gossip-protocol.png" alt="Gossip protocol" width="600">
</div>

- Each node maintains member IDs and heartbeat counters.
- Each node periodically increments its heartbeat counter.
- Each node periodically sends heartbeats to a set of random nodes.
- If the heartbeat has not increased for more than predefined periods, the member is
considered as offline

#### b. Temporary Failures
- Sloppy Quorum: Use healthy nodes to maintain operations temporarily.
<p align="center">
<img src="./images/sloppy-quorum.png" alt="Sloppy Quorum" width="400">
</p>

- After detecting failures, the system needs to deploy certain mechanisms to ensure availability
- Instead of enforcing the quorum requirement, the system chooses the first W healthy servers for writes and first R
healthy servers for reads on the hash ring.
- Offline servers are ignored. If a server is unavailable, another server will process requests temporarily

- Hinted Handoff: Offline servers catch up with changes upon recovery.
- When the down server is up, changes will be pushed back to achieve data consistency

#### c. Permanent Failures
- Use Merkle Trees for efficient synchronization between replicas.
A Merkle Tree (or hash tree) is a data structure to efficiently detect and resolve inconsistencies between replicas during permanent failures.

- Working
1. Structure:
- Leaf Nodes store the hash of individual data blocks.
- Non-Leaf Nodes store the hash of their child nodes.
- The root hash represents the combined state of all data in the tree.

2. Building a Merkle Tree:
- Step 1: Divide the key space into buckets.

<img src="./images/key-bucket.png" alt="Key Bucket" width="500">

- Step 2: Hash each key in a bucket using uniform hashing.

- Step 3: Create a single hash for each bucket.

<img src="./images/hash-bucket.png" alt="Hash Bucket" width="500">

- Step 4: Combine hashes of buckets to compute higher-level hashes, culminating in the root hash.

3. Synchronization:
- To synchronize two replicas:
- Compare their root hashes.
- If the root hashes match, the replicas are consistent.
- If the root hashes differ, compare child hashes recursively to identify inconsistent buckets.
- Only the inconsistent data is synchronized.

- Advantages
- Efficiency: Only inconsistent data is synchronized, reducing data transfer.
- Scalability: Effective for large datasets with minimal synchronization overhead.
- Reliability: Ensures data consistency across replicas.

6. Handling Data Center Outages

- Replicate data across multiple data centers to ensure availability during outages.

---

Write and Read Paths

1. Write Path (Based on Cassandra architecture)

- Persist the write in a commit log.
- Save data to a memory cache.
- Flush data to SSTable (Sorted String Table) on disk when cache is full.

2. Read Path

- Check memory cache for the data.
- If absent, use a Bloom Filter to locate the data in SSTables.
- Retrieve and return the data.

---

Final Architecture

- Clients communicate with the key-value store through simple APIs: get(key) and put(key,
value).
- A coordinator is a node that acts as a proxy between the client and the key-value store.
- Nodes are distributed on a ring using consistent hashing.
- The system is completely decentralized so adding and moving nodes can be automatic.
- Data is replicated at multiple nodes.
- There is no single point of failure as every node has the same set of responsibilities.

---

07. Unique Id Generator/Readme (07. Unique-Id Generator/Readme.md)

Chapter 7: Design a Unique ID Generator in Distributed Systems

Introduction

This chapter addresses the challenge of designing a unique ID generator for distributed systems. Traditional auto-increment keys are unsuitable in distributed environments due to scalability and synchronization challenges. The focus is on creating unique, sortable, 64-bit numerical IDs that meet the following requirements:
- IDs must be unique and ordered by date.
- IDs must fit within 64 bits.
- The system should generate over 10,000 IDs per second.

---

Step 1: Understanding the Problem

Basic Requirements

- IDs must be unique and numerical and should fit in 64 bits.
- IDs increment with time but not strictly by +1.
- IDs should be sortable by date.
- System must handle high throughput (10,000 IDs/sec).

---

Step 2: High-Level Design Options

1. Multi-Master Replication

- Approach: Use database auto_increment with step increments (e.g., +k for k servers).

- Drawbacks:
- Hard to scale across data centers.
- IDs do not always increase with time.
- Scaling issues when servers are added/removed.

2. UUID (Universally Unique Identifier)

- Approach:
- Generate 128-bit unique identifiers independently on each server using UUID.
- UUIDs can be generated independently without coordination between servers

- Advantages:
- No coordination needed between servers.
- Scales easily with web servers.
- Drawbacks:
- Exceeds 64-bit requirement.
- IDs are not sortable by time and may be non-numeric.

3. Ticket Server

- Approach: Use a centralized database server to increment and assign IDs.

- Advantages:
- Simple to implement for small-scale systems.
- Generates numeric IDs.
- Drawbacks:
- Single point of failure.
- Synchronization challenges in multi-server setups.

4. Twitter Snowflake Approach

- Approach:

- Divide IDs into sections to ensure uniqueness and scalability.
- Sign Bit (1 bit): Always 0, potentially distinguishing signed and unsigned numbers.
- Timestamp (41 bits): Milliseconds since a custom epoch (Twitter's default is 1288834974657, equivalent to Nov 04, 2010, 01:42:54 UTC). Ensures IDs are time-ordered.
- Datacenter ID (5 bits): Identifies up to 2^5 = 32 datacenters.
- Machine ID (5 bits): Identifies up to 2^5 = 32 machines within each datacenter.
- Sequence Number (12 bits): Tracks IDs generated on a machine within the same millisecond, supporting up to 2^12 = 4096 IDs per millisecond. The sequence resets to 0 every millisecond.

- Advantages:
- Scalability: Handles 10,000+ IDs per second across multiple servers.
- Time-Order: Ensures IDs are sortable by time.
- Decentralization: No single point of failure.

Step 4: Additional Considerations

1. Clock Synchronization

- Challenge: ID generation assumes synchronized clocks across servers.
- Solution: Use Network Time Protocol (NTP) to minimize drift.

2. Section Length Tuning

- Adjust section sizes (e.g., fewer sequence bits, more timestamp bits) based on use case.

3. High Availability

- ID generators are mission-critical and must be fault-tolerant.
- Consider redundancy and failover mechanisms.

---

08. URL Shortener/Readme (08. URL Shortener/Readme.md)

Chapter 8: Design a URL Shortener

Introduction

This chapter discusses the design of a URL shortening service like TinyURL. The system's main goals include URL shortening, redirecting, and high scalability to handle large traffic volumes.

Requirements

- Shortened URLs must be unique and as short as possible.
- Handle 100 million URL generations per day with a 10-year support capacity.
- Support efficient read operations with a 10:1 read-to-write ratio.
- Store 365 billion records, requiring approximately 365 TB of storage over 10 years.

---

Step 1: High-Level Design

API Endpoints

1. URL Shortening:
- Endpoint: POST api/v1/data/shorten
- Parameters: {longUrl: longURLString}
- Returns: shortURL

2. URL Redirecting:
- Endpoint: GET api/v1/shortUrl
- Returns: longURL for redirection.

URL Redirection

- 301 Redirect: A 301 redirect shows that the requested URL is “permanently” moved to the long URL. The browser caches the response, and
subsequent requests for the same URL will not be sent to the URL shortening service.
- 302 Redirect: Temporary; useful for analytics like tracking clicks.

URL Shortening

- Use a hash function to generate a short URL, mapping long URLs to unique shortened versions.
- The hash function must satisfy the following requirements:
- Each longURL must be hashed to one hashValue.
- Each hashValue can be mapped back to the longURL.

---

Step 2: Deep Dive into Design

Data Model

Store <shortURL, longURL> mappings in a relational database to optimize memory usage. The table schema includes:
- id (primary key),
- shortURL,
- longURL.

Hash Function

#### 1. Base 62 Conversion:
- Encodes numbers using characters [0-9, a-z, A-Z], providing 62 possible characters.
- Base conversion is another approach commonly used for URL shorteners.
- A unique id can be assigned to the short url and ID can be base 62 converted to get the short URL.
- A 7-character hash supports up to 3.5 trillion unique URLs, enough for 365 billion URLs.

Example:
Convert ID 2009215674938 to Base 62:
- 2009215674938 → zn9edcu.

#### 2. Hash + Collision Resolution:
- Use hash functions like CRC32, MD5, or SHA-1.

- One approach is to collect the first 7 characters of a hash value; however, this method can lead to hash collisions.
- To resolve collisions,recursively append a new predefined string until no more collision but this can be expensive.
- Resolve collisions with Bloom Filters for efficient lookup.

Comparison

- Hash + Collision Resolution:
- Fixed short URL length
- Does not need a unique ID generator
- Collision is possbile and needs resolution
- Not possible to find the next available short URL because it does not depend on ID

- Base 62 Conversion
- The length is not fixed and goes up with ID
- It needs a unique ID generator
- Collision is not possbile
- Easy to find the next short URL if ID increments by 1 (Can be a security concern)

---

URL Shortening Flow

1. Check if longURL exists in the database.
2. If found, return the existing shortURL.
3. Otherwise:
- Generate a unique ID using a distributed ID generator.
- Convert the ID to shortURL using Base 62.
- Store the <id, shortURL, longURL> mapping in the database.

---

URL Redirecting Flow

1. User clicks a shortURL.
2. Query <shortURL, longURL> mapping:
- Check the cache first for faster access.
- If not in the cache, query the database.
3. Redirect the user to longURL.

---

Additional Considerations

Rate Limiter

- Prevent abuse by setting limits on requests per IP.

Scalability

1. Web Tier: Stateless, scalable by adding/removing web servers.
2. Database Tier: Use replication and sharding.

Analytics

- Collect data like click rates, source, and timestamps for business insights.

High Availability and Reliability

- Ensure consistent and reliable services using database replication and fault-tolerant design.

---

09. Web Crawler/Readme (09. Web Crawler/Readme.md)

Chapter 9: Design a Web Crawler

Introduction

A web crawler, also known as a spider or robot, is used to discover and collect web content, such as web pages, images, and videos. This chapter focuses on designing a scalable web crawler for search engine indexing.

Applications of Web Crawlers

1. Search Engine Indexing: Collect web pages to create searchable indexes (e.g., Googlebot).
2. Web Archiving: Preserve web data for future use (e.g., US Library of Congress).
3. Web Mining: Extract knowledge from web data (e.g., financial analysis of shareholder reports).
4. Web Monitoring: Detect copyright or trademark infringements.

Design Challenges

A good web crawler must address:
- Scalability: Handle billions of pages using parallelization.
- Robustness: Manage bad HTML, crashes, and malicious links.
- Politeness: Avoid overwhelming servers with too many requests.
- Extensibility: Support new content types with minimal changes.

---

Step 1: Understanding the Problem

Requirements

1. Crawl 1 billion web pages per month (400 pages/second, peak 800 QPS).
2. Collect HTML-only content.
3. Track new and updated pages.
4. Ignore duplicate content.
5. Store crawled data for 5 years, requiring ~30 PB of storage.

---

Step 2: High-Level Design

Components

1. Seed URLs: Starting points for the crawler.
- Need to selective as a good starting point that a crawler can utilize to traverse as many links as possible.
- Can be based on locality based on different popular website or based on topics.
- Strategies: Categorize by locality or topic (e.g., sports, healthcare).

2. URL Frontier: Stores URLs to be downloaded.
- Implemented as a FIFO queue.

3. HTML Downloader: Downloads web pages from URLs provided by the URL Frontier.

4. DNS Resolver: Converts URLs to IP addresses.

5. Content Parser: Validates and parses web pages.
- Discards malformed pages.

6. Content Seen?: Checks for duplicate content using hash comparisons (compare the hash values of the two web pages).

7. Content Storage: Stores HTML pages on disk (popular content in memory to reduce latency).

8. URL Extractor: Extracts new links from parsed pages.

9. URL Filter: Excludes blacklisted or erroneous URLs.

10. URL Seen? Tracks visited URLs to avoid duplication.

11. URL Storage: Stores already visited URLs.

---

Workflow

1. Add Seed URLs to the URL Frontier.
2. HTML Downloader fetches URLs and resolves their IPs via the DNS Resolver.
3. Content Parser validates and passes content to the "Content Seen?" component.
4. If the content is new, extract links via the URL Extractor.
5. Filter and add unique links to the URL Frontier.

---

Step 3: Deep Dive into Key Components

DFS/BFS

- The web can be though of as a directed graph where web pages are nodes and hyperlinks (URLs) as edges.
- BFS is usually used for graph traversal as the depth can be be very deep thus DFS is not ideal.
- Standard BFS does not take the priority of a URL into consideration, not every page has the same level of quality and importance.

URL Frontier

- Politeness:
- Ensure only one request per host at a time. Add a dealy b/w two download tasks.
- Use a mapping from hostnames to queues and worker (download) threads.
- Each downloader thread has a separate FIFO queue and only downloads URLs from that queue.

- Queue router: Ensures that each queue (b1, b2, … bn) only contains URLs from the same host.
- Mapping table: It maps each host to a queue.
- Queue selector: Each worker thread is mapped to a FIFO queue, and it only downloads URLs from that queue. The queue selection logic is done by the Queue selector.
- Worker thread 1 to N. A worker thread downloads web pages sequentially from the same host. A delay can be added between two download tasks.

- Priority:
- Assign higher priority to important pages (e.g., by PageRank or update frequency).

<img src="./images/prioritizer.png" alt="Politeness" width="500">

- Prioritizer: It takes URLs as input and computes the priorities.
- Queue f1 to fn: Each queue has an assigned priority. Queues with high priority are selected with higher probability.
- Queue selector: Randomly choose a queue with a bias towards queues with higher priority.
- Front queues: manage prioritization
- Back queues: manage politeness

- Freshness: Recrawl based on update history or importance.

HTML Downloader

- Robots.txt Compliance: Respect rules in robots.txt files.
- Performance Optimizations:
1. Distributed crawling using multiple servers.
2. Use a DNS cache to avoid repeated lookups.
3. Geographically distribute crawl servers for faster downloads.
4. Use a short timeout to avoid slow or unresponsive servers.

Robustness

1. Consistent Hashing: Distribute load among servers effectively.
2. Error Handling: Prevent system crashes from exceptions.
3. Data Validation: Ensure content integrity.

Extensibility

- Add modules for new content types (e.g., PNG downloader, web monitor).
- Example: Plug in a module to monitor web content for copyright violations.

<img src="./images/extensibility.png" alt="Politeness" width="600">
---

Avoiding Problematic Content

1. Duplicate Content: Detect using hash comparisons.
2. Spider Traps: Avoid infinite loops with techniques like URL length limits.
3. Data Noise: Filter irrelevant content like ads or spam.

---

Step 4: Wrap Up

Key Takeaways

1. Web crawlers must balance scalability, robustness, politeness, and extensibility.
2. Politeness prevents overloading servers, while priority ensures important pages are crawled first.
3. Efficient storage and error handling are crucial for handling large-scale crawling.

Additional Considerations

- Server-Side Rendering: Handle dynamic content generated by JavaScript or AJAX.
- Anti-Spam Measures: Exclude low-quality or irrelevant pages.
- Database Sharding: Scale the data layer using replication and sharding.
- Horizontal Scaling: Use stateless servers to scale crawl jobs efficiently.
- Analytics: Collect and analyze data for insights.

---

10. Notification System/Readme (10. Notification System/Readme.md)

Chapter 10: Design a Notification System

Introduction

A notification system is essential for modern applications, providing timely updates like product notifications, events, offers, and alerts. Notifications can be sent through:
1. Push notifications (mobile or desktop),
2. SMS messages, and
3. Emails.

The chapter focuses on designing a scalable system capable of sending millions of notifications daily.

---

Step 1: Understanding the Problem

Requirements

- Notification Types: Push notifications, SMS, and Emails.
- Delivery: Soft real-time system with minimal delays.
- Platforms: iOS, Android, and desktop.
- Triggers: Notifications can be triggered by client applications or scheduled on servers.
- Scale:
- Push Notifications: 10 million/day,
- SMS: 1 million/day,
- Emails: 5 million/day.
- Opt-out Support: Users can disable specific notification types.

---

Step 2: High-Level Design

Components

1. Notification Types:
- iOS Push Notifications: Use Apple Push Notification Service (APNS).
- Android Push Notifications: Use Firebase Cloud Messaging (FCM).
- SMS Messages: Third-party services like Twilio or Nexmo.
- Emails: Commercial email services like SendGrid or Mailchimp.

2. Contact Info Gathering:
<div style="margin-left:3rem">
<img src="./images/contact-info-gathering.png" alt="Contact Info Gathering" width="500">
</div>

- Collect device tokens, phone numbers, or email addresses during app installation or signup.
- Store contact info in the database:
- Device Tokens Table: For push notifications.
- User Table: For emails and phone numbers.

3. Notification Sending Flow:

- Trigger Services:
- Generate events to initiate notifications (e.g., billing reminders, shipping updates).
- A service can be a micro-service, a cron job, or a distributed system that triggers notification sending events.
- Notification Server:
- Provide APIs for services to send notifications.
- Carry out basic validations to verify emails, phone numbers.
- Query the database or cache to fetch data needed to render a notification.
- Third-Party Services: Deliver notifications to users.

Challenges in Initial Design

- Single Point of Failure (SPOF): One notification server can crash the entire system.
- Scalability Issues: Hard to scale databases, caches, and processing components independently.
- Performance Bottlenecks: High resource demands for sending notifications.

Improved Design

- Move databases and caches out of the notification server.
- Introduce horizontal scaling with multiple notification servers.
- Use message queues to decouple system components.
- Message queues serve as buffers when high volumes of notifications are to be sent out.
- Add workers that pull notification events from message queues and send them to corresponding third party services.

---

Step 3: Design Deep Dive

Reliability

1. Prevent Data Loss:
<div style="margin-left:3rem">
<img src="./images/data-loss.png" alt="Data Loss" width="400">
</div>

- Persist notification data in a database and implement a retry mechanism.
- The Notification log database is included for data persistence.

2. Deduplication:
- Check event IDs to avoid sending duplicate notifications.
- When a notification event first arrives, check if it is seen before by checking the event ID.
If seen before discard it, otherwise send out the notification.

Additional Components

1. Notification Templates: Preformatted templates for consistent and efficient notifications.
2. Notification Settings:
- Users can opt-in or opt-out for specific channels (push, SMS, or email).
- Stored in a dedicated notification settings table.
3. Rate Limiting: Cap the frequency of notifications sent to users.
4. Retry Mechanism: Retry sending notifications if third-party services fail.
5. Monitoring Queues: Track queued notifications to scale workers dynamically.
6. Event Tracking: Collect metrics like open rate, click rate, and engagement.

Security

- Use AppKey and AppSecret to authenticate and secure APIs for push notifications.

Notification Flow

1. Trigger services call APIs to send notifications.
2. Notification servers validate requests and fetch metadata from caches or databases.
3. Notification events are sent to message queues.
4. Workers process events and interact with third-party services.
5. Third-party services deliver notifications to users.

---

Key Optimizations

1. Horizontal Scaling: Add more notification servers for load distribution.
2. Message Queues: Decouple processing to handle high volumes.
3. Caching: Reduce latency by caching frequently accessed data.
4. Distributed Crawling: Optimize message delivery geographically for better performance.

---

11. News Feed System/Readme (11. News Feed System/Readme.md)

Chapter 11: Design a News Feed System

Introduction

A news feed system displays a constantly updating list of posts (status updates, photos, videos, and links) from a user’s connections. Examples include Facebook’s news feed, Instagram’s feed, and Twitter’s timeline. This chapter explores the design of a scalable news feed system.

---

Step 1: Understanding the Problem

Requirements

1. Platform: The system supports both web and mobile apps.
2. Features:
- Users can publish posts.
- Users can view posts from friends in their news feed.
3. Sorting: Feeds are sorted in reverse chronological order for simplicity.
4. Scale:
- Users can have up to 5,000 friends.
- 10 million daily active users (DAU).
- Feeds may include text, images, and videos.

---

Step 2: High-Level Design

Overview

The design includes two main flows:
1. Feed Publishing: A user publishes a post, which is written to the database and propagated to their friends’ feeds.
2. News Feed Building: A user retrieves their news feed by aggregating posts from friends in reverse chronological order.

---

News Feed APIs

1. Feed Publishing API:
- Endpoint: POST /v1/me/feed
- Params: content (post text) and auth_token (authentication).

2. News Feed Retrieval API:
- Endpoint: GET /v1/me/feed
- Params: auth_token (authentication).

---

Feed Publishing

1. User Interaction: The user publishes a post via the feed publishing API.
2. Load Balancer: Distributes traffic to web servers.
3. Web Servers: Authenticate requests and redirect to services.
4. Post Service: Stores the post in the database and cache.
5. Fanout Service: Propagates the post to friends’ news feeds in the cache.
6. Notification Service: Sends notifications to friends.

---

News Feed Building

1. User Interaction: The user requests their news feed via the retrieval API.
2. Load Balancer: Distributes traffic to web servers.
3. Web Servers: Forward requests to the news feed service.
4. News Feed Service: Fetches post IDs from the news feed cache and retrieves complete post details from the database or cache.

---

Step 3: Design Deep Dive

Feed Publishing Deep Dive

1. Web Servers:
- Authenticate users using auth_token.
- Enforce rate limits to prevent spam.

2. Fanout Service:
- Fanout on Write: Push posts to friends’ feeds at write time.
- Pros: Real-time updates, fast feed retrieval.
- Cons: Resource-intensive for users with many friends.
- Fanout on Read: Pull posts at read time.
- Pros: Efficient for inactive users.
- Cons: Slower feed retrieval.
- Hybrid Approach: Use a push model for most users and a pull model for high-connection users (e.g., celebrities).

The fanout service works as following:

1. Fetch Friend IDs: Retrieve the friend list from a graph database.
2. Filter Friends from Cache: Access user settings in the cache to exclude certain friends (e.g., muted friends or selective sharing preferences).
3. Send to Message Queue: Send the filtered friend list along with the new post ID to a message queue for processing.
4. Fanout Workers: Workers retrieve data from the message queue and update the news feed cache. The cache stores <post_id, user_id> mappings instead of full user and post objects to save memory.
5. Store in News Feed Cache: Append new post IDs to the friends’ news feed cache. A configurable limit ensures that only recent posts are stored, as most users focus on the latest content, keeping cache memory consumption manageable.

News Feed Retrieval Deep Dive

Cache Architecture

The cache is divided into five layers:
1. News Feed Cache: Stores post IDs for quick retrieval.
2. Content Cache: Stores post details (popular posts in hot cache).
3. Social Graph Cache: Stores user relationship data.
4. Action Cache: Tracks user actions (likes, replies, shares).
5. Counter Cache: Maintains counts for likes, replies, followers, etc.

<img src="./images/cache-architecture.png" alt="Cache Architecture" width="500">
---

Key Optimizations

Scaling

1. Database Scaling:
- Horizontal scaling and sharding.
- Use of read replicas for high-traffic queries.
2. Stateless Web Tier: Keep web servers stateless to enable horizontal scaling.

Caching

1. Store frequently accessed data in memory.
2. Use cache layers to reduce latency and database load.

Reliability

1. Consistent Hashing: Distribute requests evenly across servers.
2. Message Queues: Decouple system components and buffer traffic.

Monitoring

1. Track key metrics like QPS (queries per second) and latency.
2. Monitor cache hit rates and adjust configurations accordingly.

---

12. Chat System/Readme (12. Chat System/Readme.md)

Chapter 12: Design a Chat System

Introduction

A chat system supports real-time messaging between users. This chapter focuses on designing a chat app that includes:
- One-on-One Chat
- Group Chat (max 100 users)
- Online Presence Indicators
- Multiple Device Support
- Push Notifications

The system targets 50 million daily active users (DAU) and stores chat history permanently.

---

Step 1: Understanding the Problem

Requirements

1. Features:
- One-on-one and group chat (max 100 members).
- Text-based messages (up to 100,000 characters).
- Online/offline indicators.
- Support for multiple devices.
- Push notifications.
2. Scale: Design for 50 million DAU.
3. Storage: Permanent chat history.

---

Step 2: High-Level Design

Communication Protocols

1. Sender Side: HTTP for sending messages, leveraging persistent connections for efficiency.

2. Receiver Side:
- Polling:
- Client periodically asks the server if there are messages available.
- Inefficient due to frequent, redundant requests.

- Long Polling:
- Keeps the connection open until messages arrive.
- Inefficient for inactive users.

- WebSocket:
- A bi-directional, persistent connection for real-time communication, chosen for both sending and receiving messages.
- Uses WebSockets (ws) protocol for sending and recieving messages.

<img src="./images/websocket.png" alt="Websocket" width="400" >

---

Components

1. Stateless Services:
- Handle signup, login, and user profile management.
- Integrated with service discovery to recommend the best chat server.
2. Stateful Services:
- Chat servers maintain persistent WebSocket connections.
- Responsible for message delivery and synchronization.
3. Third-Party Integration:
- Push notification services notify users about new messages.
- Refer Notification System chapter for notifications implementation.

---

Design

The client maintains a persistent WebSocket connection to a chat server for real-time messaging.

- Chat servers facilitate message sending/receiving.
- Presence servers manage online/offline status.
- API servers handle everything including user login, signup, change profile, etc.
- Notification servers send push notifications.
- Finally, the key-value store is used to store chat history.Key-value stores for the database of the chat history data for following reasons:
- It allows easy horizontal scaling.
- KV stores provide very low latency to access data.
- Relational databases do not handle long tail of data well. When the indexes grow
large, random access is expensive.
- KV stores are adopted by other proven reliable chat applications. For example,
both Facebook messenger and Discord.

Following are the data models for one-to-one chat and group chat.
- The primary key is message id, which helps to decide message sequence.
- For the group chat the composite primary key is (channel_id, message_id).
- IDs can be generated using a global 64-bit sequence number generator like Snowflake.
- A better approach is to use local sequence number generator. Local means IDs are only unique within a group.
- The reason why local IDs work is that maintaining message sequence within one-on-one channel or a group channel is sufficient.

<img src="./images/one-to-one-chat.png" alt="One to one chat design" width="300">
<img src="./images/group-chat.png" alt="Group chat design" width="300">

Step 3: Design Deep Dive

Service Discovery

- The primary role of service discovery is to recommend the best chat server for a client based
on the criteria like geographical location, server capacity.
- Uses Apache Zookeeper to allocate chat servers based on criteria like geographic location and server capacity.
- Ensures efficient load distribution and minimizes latency.

Messaging Flows

#### One-on-One Chat

1. User A sends a message to Chat Server 1.
2. Chat Server 1 assigns a unique message ID and stores the message in a key-value store.
3. If User B is online, the message is forwarded to Chat Server 2, maintaining a persistent WebSocket connection.
4. If User B is offline, a push notification is sent.

#### Group Chat

- Messages are copied to individual inboxes for each recipient in the group.
- Simplifies synchronization but becomes expensive for larger groups.
- On the recipient side, a recipient can receive messages from multiple users. Each recipient
has an inbox (message sync queue) which contains messages from different senders.

---

#### Message Synchronization

Many users have multiple devices. We need to synchronize the message across the devices.
Each device maintains a variable called cur_max_message_id, which keeps track of the latest
message ID on the device. Messages that satisfy the following two conditions are considered
as news messages:

- The recipient ID is equal to the currently logged-in user ID.
- Message ID in the key-value store is larger than cur_max_message_id

---

Online Presence

1. Heartbeat Mechanism:
<div style="margin-left:3rem">
<img src="./images/heartbeat-mechanism.png" alt="Heartbeat Mechanism" width="400">
</div>

- Clients send periodic heartbeats to presence servers to indicate they are online.
- If no heartbeat is received within a threshold (for eg x = 30), the user is marked offline.

2. Fanout Model:

- Presence updates are pushed to friends using a publish-subscribe model in which each friend pair maintains a channel.
- When User A’s online status changes, it publishes the event to three channels, channel A-B, A-C, and A-D.
- Those three channels are subscribed by User B, C, and D, respectively which get the online status updates.
- The above design is effective for a small user groups.

---

Additional Considerations

Scalability

- Horizontal Scaling: Add servers as user count increases.
- Load Balancing: Distribute traffic evenly across servers.
- Caching: Reduce database load and improve latency.

Error Handling

- Retry Mechanisms: Handle message delivery failures with retries and queuing.
- Server Failures: Use service discovery to allocate new servers in case of failures.

Future Extensions

1. Media Support: Add handling for photos and videos, including compression and cloud storage.
2. End-to-End Encryption: Ensure message privacy.
3. Client-Side Caching: Reduce data transfer for better performance.
4. Improved Load Times: Use geographically distributed caching networks.

---

13. Search Autocomplete/Readme (13. Search Autocomplete/Readme.md)

Chapter 13: Design a Search Autocomplete System

Introduction

Autocomplete, also known as typeahead or incremental search, provides real-time suggestions to users as they type in search boxes. The system must efficiently deliver top-k relevant and popular suggestions based on historical query data.

Key Features

- Suggest up to 5 autocomplete results.
- Based on query popularity (frequency).
- Support only lowercase English characters.
- Fast response time (<100 ms) and scalable.

---

Step 1: Understanding the Problem

Requirements

1. Real-Time Suggestions: Display relevant matches as the user types.
2. Top-k Results: Return up to 5 results sorted by popularity.
3. Scalability: Handle 10 million DAU with a peak QPS of 48,000.
4. High Availability: Handle failures without system downtime.
5. Data Growth: Support daily storage growth of 0.4 GB for new query data.

---

Step 2: High-Level Design

At the high-level, the system is broken down into two services:
1. Data Gathering Service:
- Collects user queries and aggregates them for frequency analysis in real-time.
- Real-time processing is not practical for large data sets; however, it is a good starting point

2. Query Service: Provides the top-k suggestions based on the user’s input.

---

Data Gathering Service

- Aggregates query data from analytics logs and updates the frequency table.
- Processes historical data weekly to build a trie (prefix tree).

Query Service

- Uses the frequency table from data gathering service.
- Processes user input and retrieves top-k suggestions from the frequency table using a Trie.
- Optimized for fast lookups using caching and efficient data structures.
- For example when a user types “tw” in the search box, the following top 5 searched queries are displayed.

---

Step 3: Design Deep Dive

Trie Data Structure

The trie is a tree-like data structure used to store and retrieve query strings efficiently.

#### Key Features
1. Compact Storage: Represents prefixes hierarchically to minimize redundancy.
2. Frequency Information: Stores the popularity of queries at each node.

4. Steps to get top k most searched queries
<div style="margin-left:3rem">
<img src="./images/trie-structure.png" alt="Trie Structure" width="500">
</div>

- Find the prefix
- Traverse the subtree from prefix node to get all valid children
- Sort the children and get top k

3. Optimizations:
- Cache top-k queries at each node to speed up retrieval and avoid traversing the whole trie.

- Limit prefix length to reduce search space as users rarely type a loong search query (say 50).

#### Trie Operations
1. Create:
- Built weekly using aggregated query data.
- The source of data is from Analytics Log/DB.
2. Update: Rarely updated in real-time; weekly updates replace old data.
3. Delete:
<div style="margin-left:3rem">
<img src="./images/delete-kv.png" alt="Delete KV" width="500">
</div>

- Filters remove unwanted or harmful suggestions (e.g., hate speech).
- Having a filter layer gives us the flexibility of removing results based on different filter rules.
- Unwanted suggestions are removed physically from the database asynchronically.

---

Query Processing Flow

1. Prefix Search:
- Identify the prefix node corresponding to the user’s input.
- Traverse the subtree to collect valid suggestions.
2. Top-k Sorting:
- Cache top-k suggestions at each node to minimize sorting overhead.
3. Response Construction:
- Construct results using cached data for fast response times.

---

Optimizations

1. Cache at Each Node:
- Store the top-k queries to avoid redundant traversals.
2. Limit Prefix Length:
- Cap prefix length to a small value (e.g., 50 characters) for faster lookups.
3. AJAX Requests:
- Use lightweight asynchronous requests for real-time responses.
4. Browser Caching:
- Save autocomplete results in the browser cache for frequently searched terms.

---

Data Gathering Pipeline

In the high-level design, whenever a user types a search query, data is updated in real-time. This appraoch is not practical.
- Users may enter billions of queries per day. Updating the trie on every query is not feasible.
- Top suggestions may not change much one the trie is built.

#### Updated Design

1. Analytics Logs:
- Stores raw query data as logs for weekly aggregation.
- Logs are append-only and are not indexed
2. Aggregators:
- Process logs into frequency tables, suitable for trie construction.
- For real-time applications such as Twitter, aggregate data in a shorter time interval.
- For other cases, aggregating data less frequently, say once per week is good enough.
3. Workers:
- Asynchronous servers rebuild the trie and store it in persistent storage.
4. Storage Options:
- Trie Cache: Trie Cache is a distributed cache system that keeps trie in memory for fast read.
- Trie DB
1. Document Store (e.g., MongoDB): Since a new trie is built weekly, we can periodically take a snapshot of it, serialize it, and store the serialized data in the database like MongoDB
2. Key-Value Store:
- Maps prefixes to node data for fast access.
- Every prefix in the trie is mapped to a key in a hash table.
- Data on each trie node is mapped to a value in a hash table.

<img src="./images/trie-db.png" alt="Trie DB" width="600">
---

Scalability

1. Sharding:
- Distribute trie nodes across servers based on prefix ranges (e.g., a-m, n-z).
- Further shard within prefixes to balance uneven distributions (e.g., aa-ag, ah-an).
2. Load Balancing:
<div style="margin-left:3rem">
<img src="./images/sharding.png" alt="Sharding" width="400">
</div>

- Use a shard map manager to route requests to the appropriate server.

---

Step 4: Advanced Features

Multi-Language Support

1. Unicode Characters: Use Unicode to support non-English languages.
2. Country-Specific Tries: Build separate tries for different countries or regions.

- Handle real-time events by dynamically updating trie nodes or weighting recent queries more heavily.

---

14. Youtube/Readme (14. Youtube/Readme.md)

Chapter 14: Design YouTube

Introduction

YouTube is a massive video streaming platform supporting video uploads, playback, and various interactions. This chapter focuses on designing a scalable video streaming system with the following core features:
- Fast video uploads
- Smooth video streaming
- Ability to change video quality
- Low infrastructure cost
- High availability and reliability

Key Statistics (2020)

- 2 billion monthly active users
- 5 billion videos watched per day
- 37% of mobile internet traffic comes from YouTube
- Available in 80 languages
- $15.1 billion ad revenue in 2019

---

Step 1: Understand the Problem and Scope

Core Functionalities

1. Upload videos
2. Watch videos

Supported Platforms

- Mobile apps, web browsers, and smart TVs

Assumptions

- Daily Active Users (DAU): 5 million
- Average Video Size: 300 MB
- Upload Limits: Max 1 GB per video
- Daily Storage Need: 150 TB
- CDN Costs: 5 million 5 videos 0.3GB * $0.02 = $150,000/day (using Amazon CloudFront)

---

Step 2: High-Level Design

Components

1. Client: Devices like smartphones, computers, and TVs.
2. CDN (Content Delivery Network): Stores and streams videos.
3. API Servers: Handles all user interactions except video streaming (e.g., uploads, metadata updates).
4. Metadata Database: Stores video metadata (e.g., title, description, size).
5. Original Storage: Blob storage for uploaded videos.
6. Transcoding Servers: Convert videos into multiple resolutions and formats.
7. Transcoded Storage: Blob storage for transcoded videos.

---

Core Workflows

#### 1. Video Uploading Flow
- Parallel Processes:
1. Upload video to original storage.
2. Update video metadata in the database.

- Video Upload (Steps):

- [1] Videos are uploaded to blob storage.
- [2] Transcoding servers convert videos to multiple formats.
- [3] One trasncoding is complete, following two steps are exectued in parallel.
- [3a] Transcoded videos are sent to transcoded storage.
- [3b] Transcoding completion events are queued in the completion queue.
- [3a.1] Videos are distributed to the CDN.
- [3b.1] Completion handlers update metadata and inform users.

- Metadata Upload (Steps):

- The client in parallel sends a request to update the video metadata
- The request contains video metadata, including file name, size, format, etc.

#### 2. Video Streaming Flow

- Videos are streamed directly from the CDN using edge servers to minimize latency.
- Some of te popular streaming protocols are MPEG_DASH, Apple HLS, Adobe HDS.
- Different streaming protocols support different video encodings and playback players.

---

Step 3: Design Deep Dive

Video Transcoding

#### Importance
1. Raw video consumes large amounts of storage space. It Reduces storage space.
2. Ensures compatibility across devices and browsers.
3. Adapts video quality to network conditions.

#### Components
- Container: Encapsulates video, audio, and metadata (e.g., MP4, AVI).
- Codecs: Compression and Decompression algorithms (e.g., H.264, VP9).

#### Directed Acyclic Graph (DAG) Model
<div style="margin-left: 3em;">
<img src="./images/dag-video-transcoding.png" alt="DAG Video Transcoding" width="600">
</div>

- Transcoding a video is computationally expensive and time-consuming.
- DAG Model defines tasks like encoding, thumbnail generation, and watermarking.
- Allows high parallelism in video processing.

- The original video is split into video, audio, and metadata.
- Video encodings: Videos are converted to support different resolutions, codec, bitrates.
- Thumbnail: It can either be uploaded by a user or automatically generated bythe system.
- Watermark: Image overlay on top of your video contains identifying information about the video.

---

Video Transcoding Architecture

1. Preprocessor: Splits videos into smaller chunks (GOP alignment). It has 4 responsibilities.

- Video splitting: Video stream is split or further split into smaller Group of Pictures (GOP) alignment.
- It split videos by GOP alignment for old clients.
- It generates DAG based on configuration files client programmers write.
- It stores GOPs and metadata in temporary storage in case the encoding fails, the system could use persisted data for retry operations.

2. DAG Scheduler: Organizes tasks into sequential or parallel stages.
<div style="margin-left: 3em;">
<img src="./images/dag-scheduler.png" alt="DAG Scheduler" width="500">
</div>

- It splits a DAG graph into stages of tasks and puts them in the task queue in the resource manager.
- Stage 1: video, audio, and metadata.
- The video file is further split into two tasks in stage 2: video encoding and thumbnail.

3. Resource Manager: Responsible for managing the efficiency of resource allocation.It
contains 3 queues and a task scheduler.
<div style="margin-left: 3em;">
<img src="./images/resource-manager.png" alt="Resource Manager" width="700">
</div>

- Task queue: priority queue that contains tasks to be executed.
- Worker queue: priority queue that contains worker utilization info.
- Running queue: contains currently running tasks and workers running the tasks.
- Task scheduler: picks the optimal task/worker, and instructs the chosen task worker to execute the job.

4. Task Workers: Perform transcoding and other operations.
<div style="margin-left: 3em;">
<img src="./images/task-worker.png" alt="Task Worker" width="250">
</div>

- Different task workers may run different tasks

5. Temporary Storage: Stores intermediate data for retries.
- The choice of storage system depends on factors like data type, data size, access frequency, data life span, etc.
6. Output: Transcoded videos ready for distribution.

---

System Optimizations

Speed Optimizations

1. Parallel Video Uploads: Split videos into smaller chunks for faster, resumable uploads.

2. Distributed Upload Centers: Use CDNs as upload hubs close to users.
3. Parallel Processing: Decouple modules using message queues for high parallelism.

Safety Optimizations

1. Pre-Signed URLs: Restrict video uploads to authorized users.

2. Protect Videos:
- DRM Systems (e.g., Apple FairPlay, Google Widevine).
- AES Encryption.
- Watermarking.

Cost-Saving Optimizations

1. Serve only popular videos via CDN; less popular ones from high-capacity servers.
2. Encode on-demand for rarely accessed videos.
3. Regionalize video distribution based on popularity.
4. Build custom CDNs and partner with ISPs to reduce bandwidth costs.

---

Error Handling

Recoverable Errors

- Retry failed uploads, transcoding, or resource allocation tasks.

Non-Recoverable Errors

- Stop malformed video processing and return error codes.

---

15. Google Drive/Readme (15. Google Drive/Readme.md)

Chapter 15: Design Google Drive

Introduction

Google Drive is a cloud-based file storage and synchronization service that allows users to store, access, and share files from various devices. This chapter discusses designing a scalable system with the following features:
- File Upload and Download
- File Sync Across Devices
- File Sharing
- File Revision History
- Notifications for Edits, Deletes, and Shares

---

Step 1: Understanding the Problem

Key Requirements

#### Functional Requirements:
- Upload and download files.
- Sync files across multiple devices.
- Maintain file revisions.
- Enable file sharing with permissions.
- Send notifications on file edits, deletions, and shares.

#### Non-Functional Requirements:
- Reliability: Data loss is unacceptable.
- Fast Sync Speed: Avoid user impatience with delayed syncing.
- Bandwidth Efficiency: Minimize unnecessary data usage.
- Scalability: Handle 10 million daily active users (DAU).
- High Availability: Operate seamlessly during server failures or network issues.

Constraints and Assumptions

- Users get 10 GB free space.
- Maximum file size: 10 GB.
- Average file upload size: 500 KB.
- Upload frequency: 2 files per day per user.
- Total storage required: 500 PB.

---

Step 2: High-Level Design

Single-Server Setup

A basic setup includes:
1. Web Server: Handles uploads and downloads.
2. Metadata Database: to keep track of metadata like user data, login info, files info/
3. Storage Directory: Holds files organized by namespaces.

- A web server and a directory called drive/ is set up as the root directory to store uploaded files.
- Under drive/ directory, there is a list of directories called namespaces.
- Each namespace contains all the uploaded files for that user.
- Each file or folder can be uniquely identified by joining the namespace and the relative path.

This design serves as a starting point but is inadequate for scaling.

#### APIs
1. Upload a file to Google Drive: Two types of uploads are supported
- Simple upload: Used when file size is small.
- Resumable upload:
- Endpoint: https://api.example.com/files/upload?uploadType=resumable
- Send the initial request to retrieve the resumable URL.
- Upload the data and monitor upload state
- If upload is disturbed, resume the upload.
2. Download a file from Google Drive: To download a file
- Endpoint: https://api.example.com/files/download
3. Get file revisions:
- Endpoint: https://api.example.com/files/list_revisions

Moving to Distributed Systems

#### Improvements:
1. Sharding: Split storage across servers based on user_id.
2. Amazon S3: Use S3 for scalable and redundant file storage with cross-region replication.

<img src="./images/replication.png" alt="Replication" width="600" />

3. Load Balancer: Distribute traffic across multiple web servers.
4. Metadata Database Replication: Ensure availability through database sharding and replication.

#### Sync Conflicts:
For a large storage system like Google Drive, sync conflicts happen from time to time.
When two users modify the same file or folder at the same time, a conflict happens.

- In the example user 1 and user 2 tries to update the same file at the same time, but user 1’s file is processed by our system first.
- User 1’s update operation goes through, but, user 2 gets a sync conflict.
- The system presents both copies of the same file: user 2’s local copy and the latest version from the server.
- User 2 has the option to merge both files or override one version with the other.

Improved design

1. User Interaction:: Users access the application via browser or mobile app.

2. Block Servers:
- Files are split into 4 MB blocks (maximum size) and assigned unique hash values.
- Blocks are stored independently in cloud storage (e.g., Amazon S3).
- File reconstruction involves joining blocks in a specific order.

3. Cloud Storage: Blocks are stored in cloud storage for scalability and redundancy.

4. Cold Storage: Inactive files are moved to cold storage to reduce costs.

5. Load Balancer: Distributes requests evenly among API servers to ensure efficient operation.

6. API Servers:
- Handle user authentication, profile management, and file metadata updates.
- Manage all non-uploading workflows.

7. Metadata Database and Cache:
- Stores metadata for users, files, blocks, and versions.
- Frequently accessed metadata is cached for faster retrieval.

8. Notification Service:
- A publisher/subscriber system that notifies clients about file changes (add, edit, delete).
- Ensures clients can pull the latest updates.

9. Offline Backup Queue: Temporarily stores file change information for offline clients to sync when back online.

---

Step 3: Design Deep Dive

Metadata Database

A highly simplified is shown below version as it only includes the most important tables and fields.
#### Schema Design:
- User Table: Stores user profiles and preferences.
- File Table: Maintains file metadata (e.g., size, name, path).
- Block Table: Tracks file blocks for reconstructing files.
- File Version Table: Stores file revision history.

---

File Upload Flow

1. File Upload:
- File is split into blocks, compressed, and encrypted by the block server.
- Blocks are uploaded to block servers and stored in S3.
2. Metadata Upload:
- Client sends metadata to the API server.
- Metadata is stored in the database with status pending.
3. Completion:
- S3 triggers a callback to update the file status to uploaded.
- Notification service informs relevant users.

---

File Sync

1. Delta Sync: Transfer only modified blocks instead of the entire file.

2. Compression: Blocks are compressed using compression algorithms depending on file types.
3. Conflict Resolution:
- First processed version wins.
- Conflicting versions are saved separately for user resolution.

---

File Download Flow

Download flow is triggered when a file is added or edited elsewhere. There are two ways a client can know:
- If client A is online while a file is changed by another client, notification service will inform client A.
- If client A is offline while a file is changed by another client, data will be saved to the cache. When the offline client is online again, it pulls the latest changes.

Once a client knows a file is changed, it first requests metadata via API servers, then
downloads blocks to construct the file.

1. Trigger: Notification service informs the client of file updates.
2. Metadata Fetch: Client retrieves updated metadata via API.
3. Block Download: Client downloads updated blocks from block servers and reconstructs the file.

---

Notification Service

1. Purpose: Keeps clients updated about file changes.
2. Mechanism: Implements long polling for asynchronous notifications.
3. Example: When a file is added, edited, or deleted, notifications are pushed to all relevant clients.

---

Storage Optimization

1. De-duplication: Remove duplicate blocks at the account level using hash-based comparisons.
2. Versioning Strategy:
- Limit the number of saved revisions.
- Prioritize recent versions for frequently edited files.
3. Cold Storage: Move rarely accessed files to cheaper storage solutions (e.g., Amazon S3 Glacier).

---

Failure Handling

1. Load Balancer Failure: Secondary load balancer becomes active.
2. Block Server Failure: Pending tasks are reassigned to other servers.
3. Metadata Database Failure:
- Promote a slave node to master.
- Redirect traffic to remaining replicas.
4. Cloud Storage Failure: Use cross-region replication to fetch unavailable files.
5. Notification Service Failure: Clients reconnect to alternative servers.

---

16. Proximity Service/Readme (16. Proximity Service/Readme.md)

Chapter 16: Proximity Service

Introduction

A proximity service is designed to find nearby locations, such as restaurants, hotels, gas stations, and other businesses. This functionality is used in applications like Google Maps and Yelp to help users discover places within a defined radius.

Step 1: Understanding the Problem and Establishing Scope

Functional Requirements

1. Search for businesses based on user location (latitude, longitude) and search radius.
2. Allow business owners to add, update, or delete businesses (not real-time).
3. Provide detailed business information when requested.

Non-Functional Requirements

- Low latency: Users should get quick responses.
- Data privacy: Compliance with GDPR and CCPA regulations.
- High availability: Handle peak-hour spikes in busy locations.

Back-of-the-Envelope Estimation

- 100 million daily active users.
- 200 million businesses in the system.
- Search QPS Calculation:
- Users make 5 searches per day.
- Search QPS = (100M × 5) / 86,400 ≈ 5,000 QPS.

---

Step 2: High-Level Design

API Design

#### Search Nearby Businesses
GET /v1/search/nearby

- Request Parameters:
- latitude: User’s location latitude.
- longitude: User’s location longitude.
- radius: Search radius (default: 5000m).

#### Business APIs
| API Endpoint | Description |
|-----------------------------------|--------------------------------------------------|
| GET /v1/businesses/{id} | Fetch detailed business info |
| POST /v1/businesses | Add a new business |
| PUT /v1/businesses/{id} | Update business details |
| DELETE /v1/businesses/{id} | Remove a business from the system |

Data Model

- Since the read volume is high because two features are very commonly used, a realtional database such as MySQL is a good fit.
- Search for nearby businesses
- View the detailed information of a business

Data Schema

- Key Database tables are the business table and the geospatial index table
- The business table consists the detailed information about a business.

High-Level System Architecture

The system comprises of two parts: Location based service (LBS) and business related service.

- Location-Based Service (LBS):
- Processes location-based search queries.
- Read-heavy service with no write requests.
- QPS is high especially during peak hours in dense areas and the system is stateless.
- Business Service: Deals with two types of requests.
- Business owners create, update or delete businesses.
- Customers view detailed information about a business.
- Load Balancer: Routes traffic to LBS and Business service.
- Database Cluster:
- Uses primary-replica architecture for read-heavy workloads.
- There might be some discrepancy between data read b/w data read by LBS and data written by by primary database.
- This incosistency is not an issue beacuase the business information is not updated in real-time.

---

Step 3: Algorithms for Fetching Nearby Businesses

Option 1: Two-Dimensional Search (Naive Approach)

The most intuitive way is to draw a circle with pre-defined radius and find all the businesses within the circle.

SQL Query:

text

SELECT business_id, latitude, longitude
FROM business
WHERE (latitude BETWEEN :lat - radius AND :lat + radius)
AND (longitude BETWEEN :long - radius AND :long + radius);

Problems:
- Inefficient: Requires scanning the entire database.
- Limited by one-dimensional indexes (latitude/longitude).

A potiential improvement is to build index on logitude and latitude columns, alhtough this is slighlty better but still vry slow.

Better Approach

- The problem with last approach is that the database index can only increase search speed in one dimension.
- An optimal apporach is to reprsent the two-dimensional data into one dimension using geospatial indexing.
- Hash: Even grid, Geo Hash
- Tree: Quadtree, Google S2, RTree

Option 2: Evenly Divided Grid

- Divides the world into fixed-size grids.
- Issue: Uneven business distribution (high density in cities, sparse in rural areas).

Option 3: Geohash

- Divide the planet into four quadrants along with the prime meridian and equator. And then divide each grid into four smaller grids.
- Each grids can be represented by altering b/w longitude and latitude bit.
- Repeat this subdivision

- Encodes latitude and longitude into a single alphanumeric string. It has 12 precisions (levels)
- Hierarchical grid structure allows for efficient searching.
- The right precision is chosen by using the minimal geohash length according to the table.
<div style="margin-left:3rem">
<img src="./images/geohash-radius-mapping.png" alt="Geohash Radius" width="400" />
</div>
- Geohash guarantees that the longer a shared prefix is between two geohashes, the closer they are.

- Challenges:
<div style="margin-left:3rem">
<img src="./images/boundary-issue.png" alt="Boundary Issue" width="300" />
</div>

- Boundary issues (businesses close to grid edges may get excluded).
- Two locations can be very close but have no shared prefix at all (can be on other side of equator)
- Two locations can have a long shared prefix but belong to different geohashes.
- Solution: Need to search neighboring grids.

Option 4: Quadtree

A quadtree is a tree data structure that recursively divides a two-dimensional space into four quadrants, with each internal node having exactly four children, representing the four sub-regions of the space.
- The quadtree is an in-memory data structure and it runs on each LBS server and built on server startup time.

- The root node is recursively broken down into 4 quadrants until no nodes are left with more than x number of businesses (100 in this case).

- The quadtree index doen't take too much memory (typically in GBs) and can easily fit in one server.
- Since tge time complexity to build the tree is nlogn, it might take a few minutes to build the tree.
- Efficient for k-nearest search queries (e.g., find the closest gas station).

#### Operational considerations
- For around 200 million businesses, it might take few minutes to build a quadtree at the server start time.
- While the quadtree is built it cannot serve traffic, therefore a new release should be rolled out incrementally to a subset of servers.
- When updating a business or adding a new the easiest approach is to incrementally rebuild the quadtree. (Leading to a lot of cache invalidation)
- Also possible to update the quadtree on the fly but more complex to implement. (Needs locking mechanism)

Option 5: Google S2

It maps a sphere to a !D index based on Hilbert curve.Two points that are close to each other on the Hilbert curve are close in 1D space.

- Divides the earth into small cells using a Hilbert curve.
- Great for geofencing becuase it can cover arbitrary areas with varying levels.
- Geofencing also allows to define parameters that surround the area of interest.
- Aother advantage if instead of having a fixed level of precision, we can specify min,max level and max cells in S2.

Tradeoff Comparison

#### Geohash
- Easy to use and implement- No need to build/rebuild a tree
- Supports fixed radius results
- Updating the index is easy.
- Cannot dynamically adjust the grid size based on population density.

#### Quadtree
- Slightly harder to implement.
- Supports fetching k-nearest businesses.
- Can dynamically adjust the grid size based on population desnsity.
- Updating the index is more complicated as might need to rebuild the whole tree.

---

Step 4: Scaling the Database and Caching Strategy

Scaling the Business Table

- Sharding by business ID ensures even data distribution.
- We have separate rows for each business in the table.

| Geohash | Business ID |
|---------|------------|
| 9q9hvu | 343 |
| 9q9hvu | 347 |
| 9q9hvu | 112 |

Scaling the Geospatial Index

- Might not be a good fit for the geohash table. In this case everything can fit in a single server so there's no tehcnical reason for sharding.
- A better approach is to have read-replicas to help with read loads.

---

Cache Strategy

The most obvious cache key choice is the location coordinate, however it has a few issues:
- Location coordinates from gps are not accurate.
- A user can move casuing the location coordinate to change.
- A better key is the geohash.

| Cache Key | Cache Value |
|------------|------------|
| geohash | List of business IDs in that grid |
| business_id | Business details (name, address, reviews, etc.) |

---

Step 5: Deployment Strategy and Final Architecture

Region and Availability Zones

- Deploy LBS and Business Service across multiple regions.

Handling Real-Time Updates

- Business updates are batch processed daily.

Final System Architecture

This final algorithm looks like this:

Steps to Retrieve Nearby Businesses

1. User Request:
- A user searches for restaurants within 500 meters.
- The client sends latitude (37.776720), longitude (-122.416730), and radius (500m) to the load balancer.

2. Request Forwarding:
- The load balancer (LB) forwards the request to the Location-Based Service (LBS).

3. Geohash Calculation:
- LBS determines the geohash length matching the radius.
- Using a reference table, 500m corresponds to geohash length = 6.

4. Fetching Neighboring Geohashes:
- LBS calculates neighboring geohashes to include nearby areas.
- The result is a list:

text

[my_geohash, neighbor1_geohash, neighbor2_geohash, ..., neighbor8_geohash]

5. Fetching Business IDs from Redis:
- For each geohash in the list, LBS queries the Geohash Redis server to fetch business IDs.
- Parallel queries are used to minimize latency.

6. Retrieving & Ranking Businesses:
- LBS fetches full business details from the Business Info Redis server.
- Businesses are sorted by distance from the user’s location.
- The ranked results are sent back to the client.

Key Optimizations

- Parallel Redis Calls: Reduces response time.
- Geohash Indexing: Ensures efficient spatial queries.
- Caching: Speeds up lookup and retrieval of business data.

This method ensures low-latency, scalable retrieval of businesses near a user’s location.

---

Choosing the Best Indexing Method

| Indexing Method | Pros | Cons |
|----------------|------|------|
| Geohash | Easy to implement, efficient for proximity search | Boundary issues, fixed grid size |
| Quadtree | Dynamically adjusts to density, supports k-nearest queries | More complex, requires tree rebalancing |
| Google S2 | Advanced geofencing, used in Google Maps | Harder to implement |

---

References

1. Geohash Algorithm
2. Quadtree Indexing
3. Google S2 Geometry

---

17. Nearby Friends/README (17. Nearby Friends/README.md)

Chapter 17: Nearby Friends

Introduction

This chapter focuses on designing a scalable backend for an application which enables user to share their location and discover friends who are nearby.

The major difference with the proximity chapter is that in this problem, locations constantly change, whereas in that one, business addresses more or less stay the same.

---

Step 1: Understand the Problem and Establish Design Scope

Some questions to drive the interview:
* C: How geographically close is considered to be "nearby"?
* I: 5 miles, this number should be configurable
* C: Is distance calculated as straight-line distance vs. taking into consideration eg a river in-between friends
* I: Yes, that is a reasonable assumption
* C: How many users does the app have?
* I: 1bil users and 10% of them use the nearby friends feature
* C: Do we need to store location history?
* I: Yes, it can be valuable for eg machine learning
* C: Can we assume inactive friends will disappear from the feature in 10min
* I: Yes
* C: Do we need to worry about GDPR, etc?
* I: No, for simlicity's sake

Functional requirements

* Users should be able to see nearby friends on their mobile app. Each friend has a distance and timestamp, indicating when the location was updated
* Nearby friends list should be updated every few seconds

Non-functional requirements

- Low latency: it's important to receive location updates without too much delay
- Reliability: Occassional data point loss is acceptable, but system should be generally available
- Eventual consistency: Location data store doesn't need strong consistency. Few seconds delay in receiving location data in different replicas is acceptable

Back-of-the-envelope

Some estimations to determine potential scale:
* Nearby friends are friends within 5mile radius
* Location refresh interval is 30s. Human walking speed is slow, hence, no need to update location too frequently.
* On average, 100mil users use the feature every day \w 10% concurrent users, ie 10mil
* On average, a user has 400 friends, all of them use the nearby friends feature
* App displays 20 nearby friends per page
* Location Update QPS = 10mil / 30 == ~334k updates per second

---

Step 2: Propose High-Level Design and Get Buy-In

Before exploring API and data model design, we'll study the communication protocol we'll use as it's less ubiquitous than traditional request-response communication model.

High-level design

At a high-level we'd want to establish effective message passing between peers. This can be done via a peer-to-peer protocol, but that's not practical for a mobile app with flaky connection and tight power consumption constraints.

A more practical approach is to use a shared backend as a fan-out mechanism towards friends you want to reach:

What does the backend do?
* Receives location updates from all active users
* For each location update, find all active users which should receive it and forward it to them
* Do not forward location data if distance between friends is beyond the configured threshold

This sounds simple but the challenge is to design the system for the scale we're operating with.

We'll start with a simpler design at first and discuss a more advanced approach in the deep dive:

- Load balancer: spreads traffic across rest API servers as well as bidirectional web socket servers
- Rest API servers: handles auxiliary tasks such as managing friends, updating profiles, etc
- Websocket servers: stateful servers, which forward location update requests to respective clients. It also manages seeding the mobile client with nearby friends locations at initialization (discussed in detail later).
- Redis location cache: used to store most recent location data for each active user. There is a TTL set on each entry in the cache. When the TTL expires, user is no longer active and their data is removed from the cache.
- User database: stores user and friendship data. Either a relational or NoSQL database can be used for this purpose.
- Location history database: stores a history of user location data, not necessarily used directly within nearby friends feature, but instead used to track historical data for analytical purposes
- Redis pubsub: used as a lightweight message bus which enables different topics for each user channel for location updates.

In the above example, websocket servers subscribe to channels for the users which are connected to them & forward location updates whenever they receive them to appropriate users.

Periodic location update

Here's how the periodic location update flow works:

* Mobile client sends a location update to the load balancer
* Load balancer forwards location update to the websocket server's persistent connection for that client
* Websocket server saves location data to location history database
* Location data is updated in location cache. Websocket server also saves location data in-memory for subsequent distance calculations for that user
* Websocket server publishes location data in user's channel via redis pub sub
* Redis pubsub broadcasts location update to all subscribers for that user channel, ie servers responsible for the friends of that user
* Subscribed web socket servers receive location update, calculate which users the update should be sent to and sends it

Here's a more detailed version of the same flow:

On average, there's going to be 40 location updates to forward as a user has 400 friends on average and 10% of them are online at a time.

API Design

Websocket Routines we'll need to support:
* periodic location update - user sends location data to websocket server
* client receives location update - server sends friend location data and timestamp
* websocket client initialization - client sends user location, server sends back nearby friends location data
* Subscribe to a new friend - websocket server sends a friend ID mobile client is supposed to track eg when friend appears online for the first time
* Unsubscribe a friend - websocket server sends a friend ID, mobile client is supposed to unsubscribe from due to eg friend going offline

HTTP API - traditional request/response payloads for auxiliary responsibilities.

Data model

* The location cache will store a mapping between user_id and lat,long,timestamp. Redis is a great choice for this cache as we only care about current location and it supports TTL eviction which we need for our use-case.
* Location history table stores the same data but in a relational table \w the four columns stated above. Cassandra can be used for this data as it is optimized for write-heavy loads.

---

Step 3: Design Deep Dive

Let's discuss how we scale the high-level design so that it works at the scale we're targetting.

How well does each component scale?

- API servers: can be easily scaled via autoscaling groups and replicating server instances
- Websocket servers: we can easily scale out the ws servers, but we need to ensure we gracefully shutdown existing connections when tearing down a server. Eg we can mark a server as "draining" in the load balancer and stop sending connections to it, prior to being finally removed from the server pool
- Client initialization: when a client first connects to a server, it fetches the user's friends, subscribes to their channels on redis pubsub, fetches their location from cache and finally forwards to client
- User database: We can shard the database based on user_id. It might also make sense to expose user/friends data via a dedicated service and API, managed by a dedicated team
- Location cache: We can shard the cache easily by spinning up several redis nodes. Also, the TTL puts a limit on the max memory we could have taken up at a time. But we still want to handle the large write load
- Redis pub/sub server: we leverage the fact that no memory is consumed if there are channels initialized but are not in use. Hence, we can pre-allocate channels for all users who use the nearby friends feature to avoid having to deal with eg bringing up a new channel when a user comes online and notifying active websocket servers

Scaling deep-dive on redis pub/sub component

We will need around 200gb of memory to maintain all pub/sub channels. This can be achieved by using 2 redis servers with 100gb each.

Given that we need to push ~14mil location updates per second, we will however need at least 140 redis servers to handle that amount of load, assuming that a single server can handle ~100k pushes per second.

Hence, we'll need a distributed redis server cluster to handle the intense CPU load.

In order to support a distributed redis cluster, we'll need to utilize a service discovery component, such as zookeeper or etcd, to keep track of which servers are alive.

What we need to encode in the service discovery component is this data:

Web socket servers use that encoded data, fetched from zookeeper to determine where a particular channel lives. For efficiency, the hash ring data can be cached in-memory on each websocket server.

In terms of scaling the server cluster up or down, we can setup a daily job to scale the cluster as needed based on historical traffic data. We can also overprovision the cluster to handle spikes in loads.

The redis cluster can be treated as a stateful storage server as there is some state maintained for the channels and there is a need for coordination with subscribers so that they hand-off to newly provisioned nodes in the cluster.

We have to be mindful of some potential issues during scaling operations:
* There will be a lot of resubscription requests from the web socket servers due to channels being moved around
* Some location updates might be missed from clients during the operation, which is acceptable for this problem, but we should still minimize it from happening. Consider doing such operation when traffic is at lowest point of the day.
* We can leverage consistent hashing to minimize amount of channels moved in the event of adding/removing servers

Adding/removing friends

Whenever a friend is added/removed, websocket server responsible for affected user needs to subscribe/unsubscribe from the friend's channel.

Since the "nearby friends" feature is part of a larger app, we can assume that a callback on the mobile client side can be registered whenever any of the events occur and the client will send a message to the websocket server to do the appropriate action.

Users with many friends

We can put a cap on the total number of friends one can have, eg facebook has a cap of 5000 max friends.

The websocket server handling the "whale" user might have a higher load on its end, but as long as we have enough web socket servers, we should be okay.

Nearby random person

What if the interviewer wants to update the design to include a feature where we can occasionally see a random person pop up on our nearby friends map?

One way to handle this is to define a pool of pubsub channels, based on geohash:

Anyone within the geohash subscribes to the appropriate channel to receive location updates for random users:

We could also subscribe to several geohashes to handle cases where someone is close but in a bordering geohash:

Alternative to Redis pub/sub

An alternative to using Redis for pub/sub is to leverage Erlang - a general programming language, optimized for distributed computing applications.

With it, we can spawn millions of small, erland processes which communicate with each other. We can handle both websocket connections and pub/sub channels within the distributed erlang application.

A challenge with using Erlang, though, is that it's a niche programming language and it could be hard to source strong erlang developers.

---

Step 4: Wrap Up

We successfully designed a system, supporting the nearby friends features.

Core components:
- Web socket servers: real-time comms between client and server
- Redis: fast read and write of location data + pub/sub channels

We also explored how to scale restful api servers, websocket servers, data layer, redis pub/sub servers and we also explored an alternative to using Redis Pub/Sub. We also explored a "random nearby person" feature.

---

18. Google Maps/README (18. Google Maps/README.md)

Chapter 18: Google Maps

Introduction

We'll design a simple version of Google Maps.

Some facts about google maps:
* Started in 2005
* Provides various services - satellite imagery, street maps, real-time traffic conditions, route planning
* By 2021, had 1bil daily active users, 99% coverage of the world, 25mil updates daily of real-time location info

---

Step 1: Understand the Problem and Establish Design Scope

Sample Q&A between candidate and interviewer:
* C: How many daily active users are we dealing with?
* I: 1bil DAU
* C: What features should we focus on?
* I: Location update, navigation, ETA, map rendering
* C: How large is road data? Do we have access to it?
* I: We obtained road data from various sources, it's TBs of raw data
* C: Should we take traffic conditions into consideration?
* I: Yes, we should for accurate time estimations
* C: How about different travel modes - by foot, biking, driving?
* I: We should support those
* C: How about multi-stop directions?
* I: Let's not focus on that for scope of interview
* C: Business places and photos?
* I: Good question, but no need to consider those

We'll focus on three key features - user location update, navigation service including ETA, map rendering.

Non-functional requirements

- Accuracy: user shouldn't get wrong directions
- Smooth navigation: Users should experience smooth map rendering
- Data and battery usage: Client should use as little data and battery as possible. Important for mobile devices.
- General availability and scalability requirements

Map 101

Before jumping into the design, there are some map-related concepts we should understand.

#### Positioning system

World is a sphere, rotating on its axis. Positiions are defined by latitude (how far north/south you are) and longitude (how far east/west you are):

#### Going from 3D to 2D

The process of translating points from 3D to 2D plane is called "map projection".

There are different ways to do it and each comes with its pros and cons. Almost all distort the actual geometry.

Google maps selected a modified version of Mercator projection called "Web Mercator".

#### Geocoding

Geocoding is the process of converting addresses to geographic coordinates.

The reverse process is called "reverse geocoding".

One way to achieve this is to use interpolation - leveraging data from different sources (eg GIS-es) where street network is mapped to geo coordinate space.

#### Geohashing

Geohashing is an encoding system which encodes a geographic area into a string of letters and digits.

It depicts the world as a flattened surface and recursively sub-divides it into four quadrants:

#### Map rendering

Map rendering happens via tiling. Instead of rendering entire map as one big custom image, world is broken up into smaller tiles.

Client only downloads relevant tiles and renders them like stitching together a mosaic.

There are different tiles for different zoom levels. Client chooses appropriate tiles based on the client's zoom level.

Eg, zooming out the entire world would download only a single 256x256 tile, representing the whole world.

#### Road data processing for navigation algorithms

In most routing algorithms, intersections are represented as nodes and roads are represented as edges:

Most navigation algorithms use a modified version of Djikstra or A* algorithms.

Pathfinding performance is sensitive to the size of the graph. To work at scale, we can't represent the whole world as a graph and run the algorithm on it.

Instead, we use a technique similar to tiling - we subdivide the world into smaller and smaller graphs.

Routing tiles hold references to neighboring tiles and algorithms can stitch together a bigger road graph as it traverses interconnected tiles:

This technique enables us to significantly reduce memory bandwidth and only load the tiles we need for the given source/destination pair.

However, for larger routes, stitching together small, detailed routing tiles would still be time/memory consuming. Instead, there are routing tiles with different level of detail and the algorithm uses the appropriately-detailed tiles, based on the destination we're headed for:

Back-of-the-envelope estimation

For storage, we need to store:
* map of the world - estimated as ~70pb based on all the tiles we need to store, but factoring in compression of very similar tiles (eg vast desert)
* metadata - negligible in size, so we can skip it from calculation
* Road info - stored as routing tiles

Estimated QPS for navigation requests - 1bil DAU at 35min of usage per week -> 5bil minutes per day.
Assuming gps update requests are batched, we arrive at 200k QPS and 1mil QPS at peak load

---

Step 2: Propose High-Level Design and Get Buy-In

Location service

It is responsible for recording a user's location updates:
* location updates are sent every t seconds
* location data streams can be used to improve the service over time, eg provide more accurate ETAs, monitor traffic data, detect closed roads, analyze user behavior, etc

Instead of sending location updates to the server all the time, we can batch the updates on the client-side and send batches instead:

Despite this optimization, for a system of Google Maps scale, load will still be significant. Therefore, we can leverage a database, optimized for heavy writes such as Cassandra.

We can also leverage Kafka for efficient stream processing of location updates, meant for further analysis.

Example location update request payload:

text

POST /v1/locations
Parameters
  locs: JSON encoded array of (latitude, longitude, timestamp) tuples.

This component is responsible for finding fast routes between A and B in a reasonable time (a little bit of latency is okay). Route need not be the fastest, but accuracy is important.

Example request payload:

text

GET /v1/nav?origin=1355+market+street,SF&destination=Disneyland

Example response:

json

{
  "distance": {"text":"0.2 mi", "value": 259},
  "duration": {"text": "1 min", "value": 83},
  "end_location": {"lat": 37.4038943, "Ing": -121.9410454},
  "html_instructions": "Head <b>northeast</b> on <b>Brandon St</b> toward <b>Lumin Way</b><div style=\"font-size:0.9em\">Restricted usage road</div>",
  "polyline": {"points": "_fhcFjbhgVuAwDsCal"},
  "start_location": {"lat": 37.4027165, "lng": -121.9435809},
  "geocoded_waypoints": [
    {
       "geocoder_status" : "OK",
       "partial_match" : true,
       "place_id" : "ChIJwZNMti1fawwRO2aVVVX2yKg",
       "types" : [ "locality", "political" ]
    },
    {
       "geocoder_status" : "OK",
       "partial_match" : true,
       "place_id" : "ChIJ3aPgQGtXawwRLYeiBMUi7bM",
       "types" : [ "locality", "political" ]
    }
  ],
  "travel_mode": "DRIVING"
}

Traffic changes and reroutes are not taken into consideration yet, those will be tackled in the deep dive section.

Map rendering

Holding the entire data set of mapping tiles on the client-side is not feasible as it's petabytes in size.

They need to be fetched on-demand from the server, based on the client's location and zoom level.

When should new tiles be fetched - while user is zooming in/out and during navigation, while they're going towards a new tile.

How should the map tiles be served to the client?
* They can be built dynamically, but that puts a huge load on the server and also makes caching hard
* Map tiles are served statically, based on their geohash, which a client can calculate. They can be statically stored & served from a CDN

CDNs enable users to fetch map tiles from point-of-presence servers (POP) which are closest to users in order to minimize latency:

Options to consider for determining map tiles:
* geohash for map tile can be calculated on the client-side. If that's the case, we should be careful that we commit to this type of map tile calculation for the long-term as forcing clients to update is hard
* alternatively, we can have simple API which calculates the map tile URLs on behalf of the clients at the cost of additional API call

---

Step 3: Design Deep Dive

Data model

Let's discuss how we store the different types of data we're dealing with.

#### Routing tiles

Initial road data set is obtained from different sources. It is improved over time based on location updates data.

The road data is unstructured. We have a periodic offline processing pipeline, which transforms this raw data into the graph-based routing tiles our app needs.

Instead of storing these tiles in a database as we don't need any database features. We can store them in S3 object storage, while caching them agressively.

We can also leverage libraries to compress adjacency lists into binary files efficiently.

#### User location data

User location data is very useful for updaring traffic conditions and doing all sorts of other analysis.

We can use Cassandra for storing this kind of data as its nature is to be write-heavy.

Example row:

#### Geocoding database

This database stores a key-value pair of lat/long pairs and places.

We can use Redis for its fast read access speed, as we have frequent read and infrequent writes.

#### Precomputed images of the world map

As we discussed, we will precompute map tiling images and store them in CDN.

Services

#### Location service

Let's focus on the database design and how user location is stored in detail for this service.

We can use a NoSQL database to facilitate the heavy write load we have on location updates. We prioritize availability over consistency as user location data often changes and becomes stale as new updates arrive.

We'll choose Cassandra as our database choice as it nicely fits all our requirements.

Example row we're going to store:

* user_id is the partition key in order to quickly access all location updates for a particular user
* timestamp is the clustering key in order to store the data sorted by the time a location update is received

We also leverage Kafka to stream location updates to various other service which need the location updates for various purposes:

#### Rendering map

Map tiles are stored at various zoom levels. At the lowest zoom level, the entire world is represented by a single 256x256 tile.

As zoom levels increase, the number of map tiles quadruples:

One optimization we can use is to not send the entire image information over the network, but instead represent tiles as vectors (paths & polygons) and let the client render the tiles dynamically.

This will have substantial bandwidth savings.

#### Navigation service

This service is responsible for finding the fastest routes:

Let's go through each component in this sub-system.

First, we have the geocoding service which resolves an address to a location of lat/long pair.

Example request:

text

https://maps.googleapis.com/maps/api/geocode/json?address=1600+Amphitheatre+Parkway,+Mountain+View,+CA

Example response:

json

{
   "results" : [
      {
         "formatted_address" : "1600 Amphitheatre Parkway, Mountain View, CA 94043, USA",
         "geometry" : {
            "location" : {
               "lat" : 37.4224764,
               "lng" : -122.0842499
            },
            "location_type" : "ROOFTOP",
            "viewport" : {
               "northeast" : {
                  "lat" : 37.4238253802915,
                  "lng" : -122.0829009197085
               },
               "southwest" : {
                  "lat" : 37.4211274197085,
                  "lng" : -122.0855988802915
               }
            }
         },
         "place_id" : "ChIJ2eUgeAK6j4ARbn5u_wAGqWA",
         "plus_code": {
            "compound_code": "CWC8+W5 Mountain View, California, United States",
            "global_code": "849VCWC8+W5"
         },
         "types" : [ "street_address" ]
      }
   ],
   "status" : "OK"
}

The route planner service computes a suggested route, optimized for travel time according to current traffic conditions.

The shortest-path service runs a variation of the A* algorithm against the routing tiles in object storage to compute an optimal path:
* It receives the source/destination pairs, converts them to lat/long pairs and derives the geohashes from those pairs to derive the routing tiles
* The algorithm starts from the initial routing tile and starts traversing it until a good enough path is found to the destination tile

The ETA service is called by the route planner to get estimated time based on machine learning algorithms, predicting ETA based on traffic data.

The ranker service is responsible to rank different possible paths based on filters, passed by the user, ie flags to avoid toll roads or freeways.

The updater service asynchronously update some of the important databases to keep them up-to-date.

#### Improvement - adaptive ETA and rerouting

One improvement we can do is to adaptively update in-flight routes based on newly available traffic data.

One way to implement this is to store users who are currently navigating through a route in the database by storing all the tiles they're supposed to go through.

Data might look like this:

text

user_1: r_1, r_2, r_3, …, r_k
user_2: r_4, r_6, r_9, …, r_n
user_3: r_2, r_8, r_9, …, r_m
...
user_n: r_2, r_10, r21, ..., r_l

If a traffic accident happens on some tile, we can identify all users whose path goes through that tile and re-route them.

To reduce the amount of tiles we store in the database, we can instead store the origin routing tile and several routing tiles in different resolution levels until the destination tile is also included:

text

user_1, r_1, super(r_1), super(super(r_1)), ...

Using this, we only need to check if the final tile of a user includes the traffic accident tile to see if user is impacted.

We can also keep track of all possible routes for a navigating user and notify them if a faster re-route is available.

#### Delivery protocols

We have several options, which enable us to proactively push data to clients from the server:
* Mobile push notifications don't work because payload is limited and it's not available for web apps
* WebSocket is generally a better option than long-polling as it has less compute footprint on servers
* We can also use server-sent events (SSE) but lean towards web sockets as they support bi-directional communication which can come in handy for eg a last-mile delivery feature

---

Step 4: Wrap Up

This is our final design:

One additional feature we could provide is multi-stop navigation which can be sold to enterprise customers such as Uber or Lyft in order to determine optimal path for visiting a set of locations.

---

19. Distributed Message Queue/README (19. Distributed Message Queue/README.md)

Chapter 19: Distributed Message Queue

Introduction

We'll be designing a distributed message queue in this chapter.

Benefits of message queues:
- Decoupling: Eliminates tight coupling between components. Let them update separately.
- Improved scalability: Producers and consumers can be scaled independently based on traffic.
- Increased availability: If one part of the system goes down, other parts continue interacting with the queue.
- Better performance: Producers can produce messages without waiting for consumer confirmation.

Some popular message queue implementations - Kafka, RabbitMQ, RocketMQ, Apache Pulsar, ActiveMQ, ZeroMQ.

Strictly speaking, Kafka and Pulsar are not message queues. They are event streaming platforms.
There is however a convergence of features which blurs the distinction between message queues and event streaming platforms.

In this chapter, we'll be building a message queue with support for more advanced features such as long data retention, repeated message consumption, etc.

---

Step 1: Understand the Problem and Establish Design Scope

Message queues ought to support few basic features - producers produce messages and consumers consume them.
There are, however, different considerations with regards to performance, message delivery, data retention, etc.

Here's a set of potential questions between Candidate and Interviewer:
* C: What's the format and average message size? Is it text only?
* I: Messages are text-only and usually a few KBs
* C: Can messages be repeatedly consumed?
* I: Yes, messages can be repeatedly consumed by different consumers. This is an added requirement, which traditional message queues don't support.
* C: Are messages consumed in the same order they were produced?
* I: Yes, order guarantee should be preserved. This is an added requirement, traditional message queues don't support this.
* C: What are the data retention requirements?
* I: Messages need to have a retention of two weeks. This is an added requirement.
* C: How many producers and consumers do we want to support?
* I: The more, the better.
* C: What data delivery semantic do we want to support? At-most-once, at-least-once, exactly-once?
* I: We definitely want to support at-least-once. Ideally, we can support all and make them configurable.
* C: What's the target throughput for end-to-end latency?
* I: It should support high throughput for use cases like log aggregation and low throughput for more traditional use cases.

Functional requirements

* Producers send messages to a message queue
* Consumers consume messages from the queue
* Messages can be consumed once or repeatedly
* Historical data can be truncated
* Message size is in the KB range
* Order of messages needs to be preserved
* Data delivery semantics is configurable - at-most-once/at-least-once/exactly-once.

Non-functional requirements

- High throughput or low latency: Configurable based on use-case
- Scalable: system should be distributed and support a sudden surge in message volume
- Persistent and durable: data should be persisted on disk and replicated among nodes

Traditional message queues typically don't support data retention and don't provide ordering guarantees. This greatly simplifies the design and we'll discuss it.

---

Step 2: Propose High-Level Design and Get Buy-In

Key components of a message queue:

* Producer sends messages to a queue
* Consumer subscribes to a queue and consumes the subscribed messages
* Message queue is a service in the middle which decouples producers from consumers, letting them scale independently.
* Producer and consumer are both clients, while the message queue is the server.

Messaging models

The first type of messaging model is point-to-point and it's commonly found in traditional message queues:

* A message is sent to a queue and it's consumed by exactly one consumer.
* There can be multiple consumers, but a message is consumed only once.
* Once message is acknowledged as consumed, it is removed from the queue.
* There is no data retention in the point-to-point model, but there is such in our design.

On the other hand, the publish-subscribe model is more common for event streaming platforms:

* In this model, messages are associated to a topic.
* Consumers are subscribed to a topic and they receive all messages sent to this topic.

Topics, partitions and brokers

What if the data volume for a topic is too large? One way to scale is by splitting a topic into partitions (aka sharding):

* Messages sent to a topic are evenly distributed across partitions
* The servers that host partitions are called brokers
* Each topic operates like a queue using FIFO for message processing. Message order is preserved within a partition.
* The position of a message within the partition is called an offset.
* Each message produced is sent to a specific partition. A partition key specifies which partition a message should land in.
* Eg a user_id can be used as a partition key to guarantee order of messages for the same user.
* Each consumer subscribes to one or more partitions. When there are multiple consumers for the same messages, they form a consumer group.

Consumer groups

Consumer groups are a set of consumers working together to consume messages from a topic:

* Messages are replicated per consumer group (not per consumer).
* Each consumer group maintains its own offset.
* Reading messages in parallel by a consumer group improves throughput but hampers the ordering guarantee.
* This can be mitigated by only allowing one consumer from a group to be subscribed to a partition.
* This means that we can't have more consumers in a group than there are partitions.

High-level architecture

- Clients: producer and consumer. Producer pushes messages to a designated topic. Consumer group subscribes to messages from a topic.
- Brokers: hold multiple partitions. A partition holds a subset of messages for a topic.
- Data storage: stores messages in partitions.
- State storage: keeps the consumer states.
- Metadata storage: stores configuration and topic properties
- Coordination service: responsible for service discovery (which brokers are alive) and leader election (which broker is leader, responsible for assigning partitions).

---

Step 3: Design Deep Dive

In order to achieve high throughput and preserve the high data retention requirement, we made some important design choices:
* We chose an on-disk data structure which takes advantage of the properties of modern HDD and disk caching strategies of modern OS-es.
* The message data structure is immutable to avoid extra copying, which we want to avoid in a high volume/high traffic system.
* We designed our writes around batching as small I/O is an enemy of high throughput.

Data storage

In order to find the best data store for messages, we must examine a message's properties:
* Write-heavy, read-heavy
* No update/delete operations. In traditional message queues, there is a "delete" operation as messages are not retained.
* Predominantly sequential read/write access pattern.

What are our options:
- Database: not ideal as typical databases don't support well both write and read heavy systems.
- Write-ahead log (WAL): a plain text file which only supports appending to it and is very HDD-friendly.
* We split partitions into segments to avoid maintaining a very large file.
* Old segments are read-only. Writes are accepted by latest segment only.

WAL files are extremely efficient when used with traditional HDDs.

There is a misconception that HDD acces is slow, but that hugely depends on the access pattern.
When the access pattern is sequential (as in our case), HDDs can achieve several MB/s write/read speed which is sufficient for our needs.
We also piggyback on the fact that the OS caches disk data in memory aggressively.

Message data structure

It is important that the message schema is compliant between producer, queue and consumer to avoid extra copying. This allows much more efficient processing.

Example message structure:

The key of the message specifies which partition a message belongs to. An example mapping is hash(key) % numPartitions.
For more flexibility, the producer can override default keys in order to control which partitions messages are distributed to.

The message value is the payload of a message. It can be plaintext or a compressed binary block.

Note: Message keys, unlike traditional KV stores, need not be unique. It is acceptable to have duplicate keys and for it to even be missing.

Other message files:
- Topic: topic the message belongs to
- Partition: The ID of the partition a message belongs to
- Offset: The position of the message in a partition. A message can be located via topic, partition, offset.
- Timestamp: When the message is stored
- Size: the size of this message
- CRC: checksum to ensure message integrity

Additional features such as filtering can be supported by adding additional fields.

Batching

Batching is critical for the performance of our system. We apply it in the producer, consumer and message queue.

It is critical because:
* It allows the operating system to group messages together, amortizing the cost of expensive network round trips
* Messages are written to the WAL in groups sequentially, which leads to a lot of sequential writes and disk caching.

There is a trade-off between latency and throughput:
* High batching leads to high throughput and higher latency.
* Less batching leads to lower throughput and lower latency.

If we need to support lower latency since the system is deployed as a traditional message queue, the system could be tuned to use a smaller batch size.

If tuned for throughput, we might need more partitions per topic to compensate for the slower sequential disk write throughput.

Producer flow

If a producer wants to send a message to a partition, which broker should it connect to?

One option is to introduce a routing layer, which route messages to the correct broker. If replication is enabled, the correct broker is the leader replica:

* Routing layer reads the replication plan from the metadata store and caches it locally.
* Producer sends a message to the routing layer.
* Message is forwarded to broker 1 who is the leader of the given partition
* Follower replicas pull the new message from the leader. Once enough confirmations are received, the leader commits the data and responds to the producer.

The reason for having replicas is to enable fault tolerance.

This approach works but has some drawbacks:
* Additional network hops due to the extra component
* The design doesn't enable batching messages

To mitigate these issues, we can embed the routing layer into the producer:

* Fewer network hops lead to lower latency
* Producers can control which partition a message is routed to
* The buffer allows us to batch messages in-memory and send out larger batches in a single request, which increases throughput.

The batch size choice is a classical trade-off between throughput and latency.

* Larger batch size leads to longer wait time before batch is committed.
* Smaller batch size leads to request being sent sooner and having lower latency but lower throughput.

Consumer flow

The consumer specifies its offset in a partition and receives a chunk of messages, beginning from that offset:

One important consideration when designing the consumer is whether to use a push or a pull model:
- Push model: leads to lower latency as broker pushes messages to consumer as it receives them.
* However, if rate of consumption falls behind the rate of production, the consumer can be overwhelmed.
* It is challenging to deal with consumers with varying processing power as the broker controls the rate of consumption.
- Pull model: leads to the consumer controlling the consumption rate.
* If rate of consumption is slow, consumer will not be overwhelmed and we can scale it to catch up.
* The pull model is more suitable for batch processing, because with the push model, the broker can't know how many messages a consumer can handle.
* With the pull model, on the other hand, consumers can aggressively fetch large message batches.
* The down side is the higher latency and extra network calls when there are no new messages. Latter issue can be mitigated using long polling.

Hence, most message queues (and us) choose the pull model.

* A new consumer subscribes to topic A and joins group 1.
* The correct broker node is found by hashing the group name. This way, all consumers in a group connect to the same broker.
* Note that this consumer group coordinator is different from the coordination service (ZooKeeper).
* Coordinator confirms that the consumer has joined the group and assigns partition 2 to that consumer.
* There are different partition assignment strategies - round-robin, range, etc.
* Consumer fetches latest messages from the last offset. The state storage keeps the consumer offsets.
* Consumer processes messages and commits the offset to the broker. The order of those operations affects the message delivery semantics.

Consumer rebalancing

Consumer rebalancing is responsible for deciding which consumers are responsible for which partition.

This process occurs when a consumer joins/leaves or a partition is added/removed.

The broker, acting as a coordinator plays a huge role in orchestrating the rebalancing workflow.

* All consumers from the same group are connected to the same coordinator. The coordinator is found by hashing the group name.
* When the consumer list changes, the coordinator chooses a new leader of the group.
* The leader of the group calculates a new partition dispatch plan and reports it back to the coordinator, which broadcasts it to the other consumers.

When the coordinator stops receiving heartbeats from the consumers in a group, a rebalancing is triggered:

Let's explore what happens when a consumer joins a group:

* Initially, only consumer A is in the group and it consumes all partitions.
* Consumer B sends a request to join the group.
* The coordinator notifies all group members that it's time to rebalance passively - as a response to the heartbeat.
* Once all consumers rejoin the group, the coordinator chooses a leader and notifies the rest about the election result.
* The leader generates the partition dispatch plan and sends it to the coordinator. Others wait for the dispatch plan.
* Consumers start consuming from the newly assigned partitions.

Here's what happens when a consumer leaves the group:

* Consumer A and B are in the same group
* Consumer B asks to leave the group
* When coordinator receives A's heartbeat, it informs them that it's time to rebalance.
* The rest of the steps are the same.

The process is similar when a consumer doesn't send a heartbeat for a long time:

State storage

The state storage stores mapping between partitions and consumers, as well as the last consumed offsets for a partition.

Group 1's offset is at 6, meaning all previous messages are consumed. If a consumer crashes, the new consumer will continue from that message on wards.

Data access patterns for consumer states:
* Frequent read/write operations, but low volume
* Data is updated frequently, but rarely deleted
* Random read/write
* Data consistency is important

Given these requirements, a fast KV storage like Zookeeper is ideal.

Metadata storage

The metadata storage stores configuration and topic properties - partition number, retention period, replica distribution.

Metadata doesn't change often and volume is small, but there is a high consistency requirement.
Zookeeper is a good choice for this storage.

ZooKeeper

Zookeeper is essential for building distributed message queues.

It is a hierarchical key-value store, commonly used for a distributed configuration, synchronization service and naming registry (ie service discovery).

With this change, the broker only needs to maintain data for the messages. Metadata and state storage is in Zookeeper.

Zookeeper also helps with leader election of the broker replicas.

Replication

In distributed systems, hardware issues are inevitable. We can tackle this via replication to achieve high availability.

* Each partition is replicated across multiple brokers, but there is only one leader replica.
* Producers send messages to leader replicas
* Followers pull the replicated messages from the leader
* Once enough replicas are synchronized, the leader returns acknowledgment to the producer
* Distribution of replicas for each partition is called the replica distribution plan.
* The leader for a given partition creates the replica distribution plan and saves it in Zookeeper

In-sync replicas

One problem we need to tackle is keeping messages in-sync between the leader and the followers for a given partition.

In-sync replicas (ISR) are replicas for a partition that stay in-sync with the leader.

The replica.lag.max.messages defines how many messages can a replica be lagging behind the leader to be considered in-sync.

* Committed offset is 13
* Two new messages are written to the leader, but not committed yet.
* A message is committed once all replicas in the ISR have synchronized that message
* Replica 2 and 3 have fully caught up with leader, hence, they are in ISR
* Replica 4 has lagged behind, hence, is removed from ISR for now

ISR reflects a trade-off between performance and durability.
* In order for producers not to lose messages, all replicas should be in sync before sending an acknowledgment
* But a slow replica will cause the whole partition to become unavailable

Acknowledgment handling is configurable.

ACK=all means that all replicas in ISR have to sync a message. Message sending is slow, but message durability is highest.

ACK=1 means that producer receives acknowledgment once leader receives the message. Message sending is fast, but message durability is low.

ACK=0 means that producer sends messages without waiting for any acknowledgment from leader. Message sending is fastest, message durability is lowest.

On the consumer side, we can connect all consumers to the leader for a partition and let them read messages from it:
* This makes for the simplest design and easiest operation
* Messages in a partition are sent to only one consumer in a group, which limits the connections to the leader replica
* The number of connections to leader replica is typically not high as long as the topic is not super hot
* We can scale a hot topic by increasing the number of partitions and consumers
* In certain scenarios, it might make sense to let a consumer lead from an ISR, eg if they're located in a separate DC

The ISR list is maintained by the leader who tracks the lag between itself and each replica.

Scalability

Let's evaluate how we can scale different parts of the system.

#### Producer

The producer is much smaller than the consumer. Its scalability can easily be achieved by adding/removing new producer instances.

#### Consumer

Consumer groups are isolated from each other. It is easy to add/remove consumer groups at will.

Rebalancing help handle the case when consumers are added/removed from a group gracefully.

Consumer groups are rebalancing help us achieve scalability and fault tolerance.

#### Broker

How do brokers handle failure?

* Once a broker fails, there are still enough replicas to avoid partition data loss
* A new leader is elected and the broker coordinator redistributes partitions which were at the failed broker to existing replicas
* Existing replicas pick up the new partitions and act as followers until they're caught up with the leader and become ISR

Additional considerations to make the broker fault-tolerant:
* The minimum number of ISRs balances latency and safety. You can fine-tune it to meet your needs.
* If all replicas of a partition are in the same node, then it's a waste of resources. Replicas should be across different brokers.
* If all replicas of a partition crash, then the data is lost forever. Spreading replicas across data centers can help, but it adds up a lot of latency. One option is to use data mirroring as a work around.

How do we handle redistribution of replicas when a new broker is added?

* We can temporarily allow more replicas than configured, until new broker catches up
* Once it does, we can remove the partition replica which is no longer needed

#### Partition

Whenever a new partition is added, the producer is notified and consumer rebalancing is triggered.

In terms of data storage, we can only store new messages to the new partition vs. trying to copy all old ones:

Decreasing the number of partitions is more involved:

* Once a partition is decommissioned, new messages are only received by remaining partitions
* The decommissioned partition isn't removed immediately as messages can still be consumed from it
* Once a pre-configured retention period passes, do we truncate the data and storage space is freed up
* During the transitional period, producers only send messages to active partitions, but consumers read from all
* Once retention period expires, consumers are rebalanced

Data delivery semantics

Let's discuss different delivery semantics.

#### At-most once

With this guarantee, messages are delivered not more than once and could not be delivered at all.

* Producer sends a message asynchronously to a topic. If message delivery fails, there is no retry.
* Consumer fetches message and immediately commits offset. If consumer crashes before processing the message, the message will not be processed.

#### At-least once

A message can be sent more than once and no message should be left unprocessed.

* Producer sends message with ack=1 or ack=all. If there is any issue, it will keep retrying.
* Consumer fetches the message and consumes the offset only after it's done processing it.
* It is possible for a message to be delivered more than once if eg consumer crashes before committing offset but after processing it.
* This is why, this is good for use-cases where data duplication is acceptable or deduplication is possible.

#### Exactly once

Extremely costly to implement for the system, albeit it's the friendliest guarantee to users:

Advanced features

Let's discuss some advanced features, we might discuss in the interview.

#### Message filtering

Some consumers might want to only consume messages of a certain type within a partition.

This can be achieved by building separate topics for each subset of messages, but this can be costly if systems have too many differing use-cases.
* It is a waste of resources to store the same message on different topics
* Producer is now tightly coupled to consumers as it changes with each new consumer requirement

We can resolve this using message filtering.
* A naive approach would be to do the filtering on the consumer-side, but that introduces unnecessary consumer traffic
* Alternatively, messages can have tags attached to them and consumers can specify which tags they're subscribed to
* Filtering could also be done via the message payloads but that can be challenging and unsafe for encrypted/serialized messages
* For more complex mathematical formulaes, the broker could implement a grammar parser or script executor, but that can be heavyweight for the message queue

#### Delayed messages & scheduled messages

For some use-cases, we might want to delay or schedule message delivery.
For example, we might submit a payment verification check for 30m from now, which triggers the consumer to see if a payment was successful.

This can be achieved by sending messages to temporary storage in the broker and moving the message to the partition at the right time:

* The temporary storage can be one or more special message topics
* The timing function can be achieved using dedicated delay queues or a hierarchical time wheel

---

Step 4: Wrap Up

Additional talking points:
- Protocol of communication: Important considerations - support all use-cases and high data volume, as well as verify message integrity. Popular protocols - AMQP and Kafka protocol.
- Retry consumption: if we can't process a message immediately, we could send it to a dedicated retry topic to be attempted later.
- Historical data archive: old messages can be backed up in high-capacity storages such as HDFS or object storage (eg S3).

---

20. Metrics Monitoring And Alerting System/README (20. Metrics Monitoring and Alerting System/README.md)

Chapter 20: Metrics Monitoring and Alerting System

Introduction

This chapter focuses on designing a highly scalable metrics monitoring and alerting system, which is critical for ensuring high availability and reliability.

---

Step 1: Understand the Problem and Establish Design Scope

A metrics monitoring system can mean a lot of different things - eg you don't want to design a logs aggregation system, when the interviewer is interested in infra metrics only.

Let's try to understand the problem first:
- C: Who are we building the system for? An in-house monitoring system for a big tech company or a SaaS like DataDog?
- I: We are building for internal use only.
- C: Which metrics do we want to collect?
- I: Operational system metrics - CPU load, Memory, Data disk space. But also high-level metrics like requests per second. Business metrics are not in scope.
- C: What is the scale of the infrastructure we're monitoring?
- I: 100mil daily active users, 1000 server pools, 100 machines per pool
- C: How long should we keep the data?
- I: Let's assume 1y retention.
- C: May we reduce metrics data resolution for long-term storage?
- I: Keep newly received metrics for 7 days. Roll them up to 1m resolution for next 30 days. Further roll them up to 1h resolution after 30 days.
- C: What are the supported alert channels?
- I: Email, phone, PagerDuty or webhooks.
- C: Do we need to collect logs such as error or access logs?
- I: No
- C: Do we need to support distributed system tracing?
- I: No

High-level requirements and assumptions

The infrastructure being monitored is large-scale:
- 100mil DAU
- 1000 server pools 100 machines ~100 metrics per machine -> ~10mil metrics
- 1-year data retention
- Data retention policy - raw for 7d, 1-minute resolution for 30d, 1h resolution for 1y

A variety of metrics can be monitored:
- CPU load
- Request count
- Memory usage
- Message count in message queues

Non-functional requirements

- Scalability: System should be scalable to accommodate more metrics and alerts
- Low latency: System needs to have low query latency for dashboards and alerts
- Reliability: System should be highly reliable to avoid missing critical alerts
- Flexibility: System should be able to easily integrate new technologies in the future

What requirements are out of scope?
- Log monitoring: the ELK stack is very popular for this use-case
- Distributed system tracing: this refers to collecting data about a request lifecycle as it flows through multiple services within the system

---

Step 2: Propose High-Level Design and Get Buy-In

Fundamentals

There are five core components involved in a metrics monitoring and alerting system:

- Data collection: collect metrics data from different sources
- Data transmission: transfer data from sources to the metrics monitoring system
- Data storage: organize and store incoming data
- Alerting: Analyze incoming data, detect anomalies and generate alerts
- Visualization: Present data in graphs, charts, etc

Data model

Metrics data is usually recorded as a time-series, which contains a set of values with timestamps.
The series can be identified by name and an optional set of tags.

Example 1 - What is the CPU load on production server instance i631 at 20:00?

The data can be identified by the following table:

The time series is identified by the metric name, labels and a single point in at a specific time.

Example 2 - What is the average CPU load across all web servers in the us-west region for the last 10min?

text

CPU.load host=webserver01,region=us-west 1613707265 50
CPU.load host=webserver01,region=us-west 1613707265 62
CPU.load host=webserver02,region=us-west 1613707265 43
CPU.load host=webserver02,region=us-west 1613707265 53
...
CPU.load host=webserver01,region=us-west 1613707265 76CPU.load host=webserver01,region=us-west 1613707265 83

This is an example data we might pull from storage to answer that question.
The average CPU load can be calculated by averaging the values in the last column of the rows.

The format shown above is called the line protocol and is used by many popular monitoring software in the market - eg Prometheus, OpenTSDB.

What every time series consists of:

A good way to visualize how data looks like:

- The x axis is the time
- the y axis is the dimension you're querying - eg metric name, tag, etc.

The data access pattern is write-heavy and spiky reads as we collect a lot of metrics, but they are infrequently accessed, although in bursts when eg there are ongoing incidents.

The data storage system is the heart of this design.
- It is not recommended to use a general-purpose database for this problem, although you could achieve good scale \w expert-level tuning.
- Using a NoSQL database can work in theory, but it is hard to devise a scalable schema for effectively storing and querying time-series data.

There are many databases, specifically tailored for storing time-series data. Many of them support custom query interfaces which allow for effective querying of time-series data.
- OpenTSDB is a distributed time-series database, but it is based on Hadoop and HBase. If you don't have that infrastructure provisioned, it would be hard to use this tech.
- Twitter uses MetricsDB, while Amazon offers Timestream.
- The two most popular time-series databases are InfluxDB and Prometheus.
- They are designed to store large volumes of time-series data. Both of them are based on in-memory cache + on-disk storage.

Example scale of InfluxDB - more than 250k writes per second when provisioned with 8 cores and 32gb RAM:

It is not expected for you to understand the internals of a metrics database as it is niche knowledge. You might be asked only if you've mentioned it on your resume.

For the purposes of the interview, it is sufficient to understand that metrics are time-series data and to be aware of popular time-series databases, like InfluxDB.

One nice feature of time-series databases is the efficient aggregation and analysis of large amounts of time-series data by labels.
InfluxDB, for example, builds indexes for each label.

It is critical, however, to keep the cardinality of labels low - ie, not using too many unique labels.

High-level Design

- Metrics source: can be application servers, SQL databases, message queues, etc.
- Metrics collector: Gathers metrics data and writes to time-series database
- Time-series database: stores metrics as time-series. Provides a custom query interface for analyzing large amounts of metrics.
- Query service: Makes it easy to query and retrieve data from the time-series DB. Could be replaced entirely by the DB's interface if it's sufficiently powerful.
- Alerting system: Sends alert notifications to various alerting destinations.
- Visualization system: Shows metrics in the form of graphs/charts.

---

Step 3: Design Deep Dive

Let's deep dive into several of the more interesting parts of the system.

Metrics collection

For metrics collection, occasional data loss is not critical. It's acceptable for clients to fire and forget.

There are two ways to implement metrics collection - pull or push.

Here's how the pull model might look like:

For this solution, the metrics collector needs to maintain an up-to-date list of services and metrics endpoints.
We can use Zookeeper or etcd for that purpose - service discovery.

Service discovery contains contains configuration rules about when and where to collect metrics from:

Here's a detailed explanation of the metrics collection flow:

- Metrics collector fetches configuration metadata from service discovery. This includes pulling interval, IP addresses, timeout & retry params.
- Metrics collector pulls metrics data via a pre-defined http endpoint (eg /metrics). This is typically done by a client library.
- Alternatively, the metrics collector can register a change event notification with the service discovery to be notified once the service endpoint changes.
- Another option is for the metrics collector to periodically poll for metrics endpoint configuration changes.

At our scale, a single metrics collector is not enough. There must be multiple instances.
However, there must also be some kind of synchronization among them so that two collectors don't collect the same metrics twice.

One solution for this is to position collectors and servers on a consistent hash ring and associate a set of servers with a single collector only:

With the push model, on the other hand, services push their metrics to the metrics collector proactively:

In this approach, typically a collection agent is installed alongside service instances.
The agent collects metrics from the server and pushes them to the metrics collector.

With this model, we can potentially aggregate metrics before sending them to the collector, which reduces the volume of data processed by the collector.

On the flip side, metrics collector can reject push requests as it can't handle the load.
It is important, hence, to add the collector to an auto-scaling group behind a load balancer.

so which one is better? There are trade-offs between both approaches and different systems use different approaches:
- Prometheus uses a pull architecture
- Amazon Cloud Watch and Graphite use a push architecture

Here are some of the main differences between push and pull:
| | Pull | Push |
|----------------------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| Easy debugging | The /metrics endpoint on application servers used for pulling metrics can be used to view metrics at any time. You can even do this on your laptop. Pull wins. | If the metrics collector doesn't receive metrics, the problem might be caused by network issues. |
| Health check | If an application server doesn't respond to the pull, you can quickly figure out if an application server is down. Pull wins. | If the metrics collector doesn't receive metrics, the problem might be caused by network issues. |
| Short-lived jobs | | Some of the batch jobs might be short-lived and don't last long enough to be pulled. Push wins. This can be fixed by introducing push gateways for the pull model [22]. |
| Firewall or complicated network setups | Having servers pulling metrics requires all metric endpoints to be reachable. This is potentially problematic in multiple data center setups. It might require a more elaborate network infrastructure. | If the metrics collector is set up with a load balancer and an auto-scaling group, it is possible to receive data from anywhere. Push wins. |
| Performance | Pull methods typically use TCP. | Push methods typically use UDP. This means the push method provides lower-latency transports of metrics. The counterargument here is that the effort of establishing a TCP connection is small compared to sending the metrics payload. |
| Data authenticity | Application servers to collect metrics from are defined in config files in advance. Metrics gathered from those servers are guaranteed to be authentic. | Any kind of client can push metrics to the metrics collector. This can be fixed by whitelisting servers from which to accept metrics, or by requiring authentication. |

There is no clear winner. A large organization probably needs to support both. There might not be a way to install a push agent in the first place.

Scale the metrics transmission pipeline

The metrics collector is provisioned in an auto-scaling group, regardless if we use the push or pull model.

There is a chance of data loss if the time-series DB is down, however. To mitigate this, we'll provision a queuing mechanism:

- Metrics collectors push metrics data into kafka
- Consumers or stream processing services such as Apache Storm, Flink or Spark process the data and push it to the time-series DB

This approach has several advantages:
- Kafka is used as a highly-reliable and scalable distributed message platform
- It decouples data collection and data processing from one another
- It can prevent data loss by retaining the data in Kafka

Kafka can be configured with one partition per metric name, so that consumers can aggregate data by metric names.
To scale this, we can further partition by tags/labels and categorize/prioritize metrics to be collected first.

The main downside of using Kafka for this problem is the maintenance/operation overhead.
An alternative is to use a large-scale ingestion system like Gorilla.
It can be argued that using that would be as scalable as using Kafka for queuing.

Where aggregations can happen

Metrics can be aggregated at several places. There are trade-offs between different choices:
- Collection agent: client-side collection agent only supports simple aggregation logic. Eg collect a counter for 1m and send it to the metrics collector.
- Ingestion pipeline: To aggregate data before writing to the DB, we need a stream processing engine like Flink. This reduces write volume, but we lose data precision as we don't store raw data.
- Query side: We can aggregate data when we run queries via our visualization system. There is no data loss, but queries can be slow due to a lot of data processing.

Query Service

Having a separate query service from the time-series DB decouples the visualization and alerting system from the database, which enables us to decouple the DB from clients and change it at will.

We can add a Cache layer here to reduce the load to the time-series database:

We can also avoid adding a query service altogether as most visualization and alerting systems have powerful plugins to integrate with most time-series databases.
With a well-chosen time-series DB, we might not need to introduce our own caching layer as well.

Most time-series DBs don't support SQL simply because it is ineffective for querying time-series data. Here's an example SQL query for computing an exponential moving average:

text

select id,
       temp,
       avg(temp) over (partition by group_nr order by time_read) as rolling_avg
from (
  select id,
         temp,
         time_read,
         interval_group,
         id - row_number() over (partition by interval_group order by time_read) as group_nr
  from (
    select id,
    time_read,
    "epoch"::timestamp + "900 seconds"::interval * (extract(epoch from time_read)::int4 / 900) as interval_group,
    temp
    from readings
  ) t1
) t2
order by time_read;

Here's the same query in Flux - query language used in InfluxDB:

text

from(db:"telegraf")
  |> range(start:-1h)
  |> filter(fn: (r) => r._measurement == "foo")
  |> exponentialMovingAverage(size:-10s)

Storage layer

It is important to choose the time-series database carefully.

According to research published by Facebook, ~85% of queries to the operational store were for data from the past 26h.

If we choose a database, which harnesses this property, it could have significant impact on system performance. InfluxDB is one such option.

Regardless of the database we choose, there are some optimizations we might employ.

Data encoding and compression can significantly reduce the size of data. Those features are usually built into a good time-series database.

In the above example, instead of storing full timestamps, we can store timestamp deltas.

Another technique we can employ is down-sampling - converting high-resolution data to low-resolution in order to reduce disk usage.

We can use that for old data and make the rules configurable by data scientists, eg:
- 7d - no down-sampling
- 30d - down-sample to 1min
- 1y - down-sample to 1h

For example, here's a 10-second resolution metrics table:
| metric | timestamp | hostname | Metric_value |
|--------|----------------------|----------|--------------|
| cpu | 2021-10-24T19:00:00Z | host-a | 10 |
| cpu | 2021-10-24T19:00:10Z | host-a | 16 |
| cpu | 2021-10-24T19:00:20Z | host-a | 20 |
| cpu | 2021-10-24T19:00:30Z | host-a | 30 |
| cpu | 2021-10-24T19:00:40Z | host-a | 20 |
| cpu | 2021-10-24T19:00:50Z | host-a | 30 |

down-sampled to 30-second resolution:
| metric | timestamp | hostname | Metric_value (avg) |
|--------|----------------------|----------|--------------------|
| cpu | 2021-10-24T19:00:00Z | host-a | 19 |
| cpu | 2021-10-24T19:00:30Z | host-a | 25 |

Finally, we can also use cold storage to use old data, which is no longer used. The financial cost for cold storage is much lower.

Alerting system

Configuration is loaded to cache servers. Rules are typically defined in YAML format. Here's an example:

text

- name: instance_down rules:

# Alert for any instance that is unreachable for >5 minutes. - alert: instance_down expr: up == 0 for: 5m labels: severity: page

The alert manager fetches alert configurations from cache. Based on configuration rules, it also calls the query service at a predefined interval.
If a rule is met, an alert event is created.

Other responsibilities of the alert manager are:
- Filtering, merging and deduplicating alerts. Eg if an alert of a single instance is triggered multiple times, only one alert event is generated.
- Access control - it is important to restrict alert-management operations to certain individuals only
- Retry - the manager ensures that the alert is propagated at least once.

The alert store is a key-value database, like Cassandra, which keeps the state of all alerts. It ensures a notification is sent at least once.
Once an alert is triggered, it is published to Kafka.

Finally, alert consumers pull alerts data from Kafka and send notifications over to different channels - Email, text message, PagerDuty, webhooks.

In the real-world, there are many off-the-shelf solutions for alerting systems. It is difficult to justify building your own system in-house.

Visualization system

The visualization system shows metrics and alerts over a time period. Here's an dashboard built with Grafana:

A high-quality visualization system is very hard to build. It is hard to justify not using an off-the-shelf solution like Grafana.

---

Step 4: Wrap up

Here's our final design:

---

21. Ad Click Event Aggregation/README (21. Ad Click Event Aggregation/README.md)

Chapter 21: Ad Click Event Aggregation

Introduction

Digital advertising is a big industry with the rise of Facebook, YouTube, TikTok, etc.

Hence, tracking ad click events is important. In this chapter, we explore how to design an ad click event aggregation system at Facebook/Google scale.

Digital advertising has a process called real-time bidding (RTB), where digital advertising inventory is bought and sold:

Speed of RTB is important as it usually occurs within a second.
Data accuracy is also very important as it impacts how much money advertisers pay.

Based on ad click event aggregations, advertisers can make decisions such as adjust target audience and keywords.

---

Step 1: Understand the Problem and Establish Design Scope

- C: What is the format of the input data?
- I: 1bil ad clicks per day and 2mil ads in total. Number of ad-click events grows 30% year-over-year.
- C: What are some of the most important queries our system needs to support?
- I: Top queries to take into consideration:
- Return number of click events for ad X in last Y minutes
- Return top 100 most clicked ads in the past 1min. Both parameters should be configurable. Aggregation occurs every minute.
- Support data filtering by ip, user_id, country for the above queries
- C: Do we need to worry about edge cases? Some of the ones I can think of:
- There might be events that arrive later than expected
- There might be duplicate events
- Different parts of the system might be down, so we need to consider system recovery
- I: That's a good list, take those into consideration
- C: What is the latency requirement?
- I: A few minutes of e2e latency for ad click aggregation. For RTB, it is less than a second. It is ok to have that latency for ad click aggregation as those are usually used for billing and reporting.

Functional requirements

- Aggregate the number of clicks of ad_id in the last Y minutes
- Return top 100 most clicked ad_id every minute
- Support aggregation filtering by different attributes
- Dataset volume is at Facebook or Google scale

Non-functional requirements

- Correctness of the aggregation result is important as it's used for RTB and ads billing
- Properly handle delayed or duplicate events
- Robustness - system should be resilient to partial failures
- Latency - a few minutes of e2e latency at most

Back-of-the-envelope estimation

- 1bil DAU
- Assuming user clicks 1 ad per day -> 1bil ad clicks per day
- Ad click QPS = 10,000
- Peak QPS is 5 times the number = 50,000
- A single ad click occupies 0.1KB storage. Daily storage requirement is 100gb
- Monthly storage = 3tb

---

Step 2: Propose High-Level Design and Get Buy-In

In this section, we discuss query API design, data model and high-level design.

Query API Design

The API is a contract between the client and the server. In our case, the client is the dashboard user - data scientist/analyst, advertiser, etc.

Here's our functional requirements:
- Aggregate the number of clicks of ad_id in the last Y minutes
- Return top N most clicked ad_id in the last M minutes
- Support aggregation filtering by different attributes

We need two endpoints to achieve those requirements. Filtering can be done via query parameters on one of them.

Aggregate number of clicks of ad_id in the last M minutes:

text

GET /v1/ads/{:ad_id}/aggregated_count

Query parameters:
- from - start minute. Default is now - 1 min
- to - end minute. Default is now
- filter - identifier for different filtering strategies. Eg 001 means "non-US clicks".

Response:
- ad_id - ad identifier
- count - aggregated count between start and end minutes

Return top N most clicked ad_ids in the last M minutes

text

GET /v1/ads/popular_ads

Query parameters:
- count - top N most clicked ads
- window - aggregation window size in minutes
- filter - identifier for different filtering strategies

Response:
- list of ad_ids

Data model

In our system, we have raw and aggregated data.

Raw data looks like this:

text

[AdClickEvent] ad001, 2021-01-01 00:00:01, user 1, 207.148.22.22, USA

Here's an example in a structured format:
| ad_id | click_timestamp | user | ip | country |
|-------|---------------------|-------|---------------|---------|
| ad001 | 2021-01-01 00:00:01 | user1 | 207.148.22.22 | USA |
| ad001 | 2021-01-01 00:00:02 | user1 | 207.148.22.22 | USA |
| ad002 | 2021-01-01 00:00:02 | user2 | 209.153.56.11 | USA |

Here's the aggregated version:
| ad_id | click_minute | filter_id | count |
|-------|--------------|-----------|-------|
| ad001 | 202101010000 | 0012 | 2 |
| ad001 | 202101010000 | 0023 | 3 |
| ad001 | 202101010001 | 0012 | 1 |
| ad001 | 202101010001 | 0023 | 6 |

The filter_id helps us achieve our filtering requirements.
| filter_id | region | IP | user_id |
|-----------|--------|-----------|---------|
| 0012 | US | | |
| 0013 | | 123.1.2.3 | |

To support quickly returning top N most clicked ads in the last M minutes, we'll also maintain this structure:
| most_clicked_ads | | |
|--------------------|-----------|--------------------------------------------------|
| window_size | integer | The aggregation window size (M) in minutes |
| update_time_minute | timestamp | Last updated timestamp (in 1-minute granularity) |
| most_clicked_ads | array | List of ad IDs in JSON format. |

What are some pros and cons between storing raw data and storing aggregated data?
- Raw data enables using the full data set and supports data filtering and recalculation
- On the other hand, aggregated data allows us to have a smaller data set and a faster query
- Raw data means having a larger data store and a slower query
- Aggregated data, however, is derived data, hence there is some data loss.

In our design, we'll use a combination of both approaches:
- It's a good idea to keep the raw data around for debugging. If there is some bug in aggregation, we can discover the bug and backfill.
- Aggregated data should be stored as well for faster query performance.
- Raw data can be stored in cold storage to avoid extra storage costs.

When it comes to the database, there are several factors to take into consideration:
- What does the data look like? Is it relational, document or blob?
- Is the workload read-heavy, write-heavy or both?
- Are transactions needed?
- Do the queries rely on OLAP functions like SUM and COUNT?

For the raw data, we can see that the average QPS is 10k and peak QPS is 50k, so the system is write-heavy.
On the other hand, read traffic is low as raw data is mostly used as backup if anything goes wrong.

Relational databases can do the job, but it can be challenging to scale the writes.
Alternatively, we can use Cassandra or InfluxDB which have better native support for heavy write loads.

Another option is to use Amazon S3 with a columnar data format like ORC, Parquet or AVRO. Since this setup is unfamiliar, we'll stick to Cassandra.

For aggregated data, the workload is both read and write heavy as aggregated data is constantly queried for dashboards and alerts.
It is also write-heavy as data is aggregated and written every minute by the aggregation service.
Hence, we'll use the same data store (Cassandra) here as well.

High-level design

Here's how our system looks like:

Data flows as an unbounded data stream on both inputs and outputs.

In order to avoid having a synchronous sink, where a consumer crashing can cause the whole system to stall,
we'll leverage asynchronous processing using message queues (Kafka) to decouple consumers and producers.

The first message queue stores ad click event data:
| ad_id | click_timestamp | user_id | ip | country |
|-------|-----------------|---------|----|---------|

The second message queue contains ad click counts, aggregated per-minute:
| ad_id | click_minute | count |
|-------|--------------|-------|

As well as top N clicked ads aggregated per minute:
| update_time_minute | most_clicked_ads |
|--------------------|------------------|

The second message queue is there in order to achieve end to end exactly-once atomic commit semantics:

For the aggregation service, using the MapReduce framework is a good option:

Each node is responsible for one single task and it sends the processing result to the downstream node.

The map node is responsible for reading from the data source, then filtering and transforming the data.

For example, the map node can allocate data across different aggregation nodes based on the ad_id:

Alternatively, we can distribute ads across Kafka partitions and let the aggregation nodes subscribe directly within a consumer group.
However, the mapping node enables us to sanitize or transform the data before subsequent processing.

Another reason might be that we don't have control over how data is produced,
so events related to the same ad_id might go on different partitions.

The aggregate node counts ad click events by ad_id in-memory every minute.

The reduce node collects aggregated results from aggregate node and produces the final result:

This DAG model uses the MapReduce paradigm. It takes big data and leverages parallel distributed computing to turn it into regular-sized data.

In the DAG model, intermediate data is stored in-memory and different nodes communicate with each other using TCP or shared memory.

Let's explore how this model can now help us to achieve our various use-cases.

Use-case 1 - aggregate the number of clicks:

- Ads are partitioned using ad_id % 3

Use-case 2 - return top N most clicked ads:

- In this case, we're aggregating the top 3 ads, but this can be extended to top N ads easily
- Each node maintains a heap data structure for fast retrieval of top N ads

Use-case 3 - data filtering:
To support fast data filtering, we can predefine filtering criterias and pre-aggregate based on it:
| ad_id | click_minute | country | count |
|-------|--------------|---------|-------|
| ad001 | 202101010001 | USA | 100 |
| ad001 | 202101010001 | GPB | 200 |
| ad001 | 202101010001 | others | 3000 |
| ad002 | 202101010001 | USA | 10 |
| ad002 | 202101010001 | GPB | 25 |
| ad002 | 202101010001 | others | 12 |

This technique is called the star schema and is widely used in data warehouses.
The filtering fields are called dimensions.

This approach has the following benefits:
- Simple to undertand and build
- Current aggregation service can be reused to create more dimensions in the star schema.
- Accessing data based on filtering criteria is fast as results are pre-calculated

A limitation of this approach is that it creates many more buckets and records, especially when we have lots of filtering criterias.

---

Step 3: Design Deep Dive

Let's dive deeper into some of the more interesting topics.

Streaming vs. Batching

The high-level architecture we proposed is a type of stream processing system.
Here's a comparison between three types of systems:
| | Services (Online system) | Batch system (offline system) | Streaming system (near real-time system) |
|-------------------------|-------------------------------|--------------------------------------------------------|----------------------------------------------|
| Responsiveness | Respond to the client quickly | No response to the client needed | No response to the client needed |
| Input | User requests | Bounded input with finite size. A large amount of data | Input has no boundary (infinite streams) |
| Output | Responses to clients | Materialized views, aggregated metrics, etc. | Materialized views, aggregated metrics, etc. |
| Performance measurement | Availability, latency | Throughput | Throughput, latency |
| Example | Online shopping | MapReduce | Flink [13] |

In our design, we used a mixture of batching and streaming.

We used streaming for processing data as it arrives and generates aggregated results in near real-time.
We used batching, on the other hand, for historical data backup.

A system which contains two processing paths - batch and streaming, simultaneously, this architecture is called lambda.
A disadvantage is that you have two processing paths with two different codebases to maintain.

Kappa is an alternative architecture, which combines batch and stream processing in one processing path.
The key idea is to use a single stream processing engine.

Lambda architecture:

Kappa architecture:

Our high-level design uses Kappa architecture as reprocessing of historical data also goes through the aggregation service.

Whenever we have to recalculate aggregated data due to eg a major bug in aggregation logic, we can recalculate the aggregation from the raw data we store.
- Recalculation service retrieves data from raw storage. This is a batch job.
- Retrieved data is sent to a dedicated aggregation service, so that the real-time processing aggregation service is not impacted.
- Aggregated results are sent to the second message queue, after which we update the results in the aggregation database.

Time

We need a timestamp to perform aggregation. It can be generated in two places:
- event time - when ad click occurs
- Processing time - system time when the server processes the event

Due to the usage of async processing (message queues) and network delays, there can be significant difference between event time and processing time.
- If we use processing time, aggregation results can be inaccurate
- If we use event time, we have to deal with delayed events

There is no perfect solution, we need to consider trade-offs:
| | Pros | Cons |
|-----------------|---------------------------------------|--------------------------------------------------------------------------------------|
| Event time | Aggregation results are more accurate | Clients might have the wrong time or timestamp might be generated by malicious users |
| Processing time | Server timestamp is more reliable | The timestamp is not accurate if event is late |

Since data accuracy is important, we'll use the event time for aggregation.

To mitigate the issue of delayed events, a technique called "watermark" can be leveraged.

In the example below, event 2 misses the window where it needs to be aggregated:

However, if we purposefully extend the aggregation window, we can reduce the likelihood of missed events.
The extended part of a window is called a "watermark":

- Short watermark increases likelihood of missed events, but reduces latency
- Longer watermark reduces likelihood of missed events, but increases latency

There is always likelihood of missed events, regardless of the watermark's size. But there is no use in optimizing for such low-probability events.

We can instead resolve such inconsistencies by doing end-of-day reconciliation.

Aggregation window

There are four types of window functions:
- Tumbling (fixed) window
- Hopping window
- Sliding window
- Session window

In our design, we leverage a tumbling window for ad click aggregations:

As well as a sliding window for the top N clicked ads in M minutes aggregation:

Delivery guarantees

Since the data we're aggregating is going to be used for billing, data accuracy is a priority.

Hence, we need to discuss:
- How to avoid processing duplicate events
- How to ensure all events are processed

There are three delivery guarantees we can use - at-most-once, at-least-once and exactly once.

In most circumstances, at-least-once is sufficient when a small amount of duplicates is acceptable.
This is not the case for our system, though, as a difference in small percent can result in millions of dollars of discrepancy.
Hence, we'll need to use exactly-once delivery semantics.

Data deduplication

One of the most common data quality issues is duplicated data.

It can come from a wide range of sources:
- Client-side - a client might resend the same event multiple times. Duplicated events sent with malicious intent are best handled by a risk engine.
- Server outage - An aggregation service node goes down in the middle of aggregation and the upstream service hasn't received an acknowledgment so event is resent.

Here's an example of data duplication occurring due to failure to acknowledge an event on the last hop:

In this example, offset 100 will be processed and sent downstream multiple times.

One option to try and mitigate this is to store the last seen offset in HDFS/S3, but this risks the result never reaching downstream:

Finally, we can store the offset while interacting with downstream atomically. To achieve this, we need to implement a distributed transaction:

Personal side-note: Alternatively, if the downstream system handles the aggregation result idempotently, there is no need for a distributed transaction.

Scale the system

Let's discuss how we scale the system as it grows.

We have three independent components - message queue, aggregation service and database.
Since they are decoupled, we can scale them independently.

How do we scale the message queue:
- We don't put a limit on producers, so they can be scaled easily
- Consumers can be scaled by assigning them to consumer groups and increasing the number of consumers.
- For this to work, we also need to ensure there are enough partitions created preemptively
- Also, consumer rebalancing can take a while when there are thousands of consumers so it is recommended to do it off peak hours
- We could also consider partitioning the topic by geography, eg topic_na, topic_eu, etc.

How do we scale the aggregation service:

- The map-reduce nodes can easily be scaled by adding more nodes
- The throughput of the aggregation service can be scaled by by utilising multi-threading
- Alternatively, we can leverage resource providers such as Apache YARN to utilize multi-processing
- Option 1 is easier, but option 2 is more widely used in practice as it's more scalable
- Here's the multi-threading example:

How do we scale the database:
- If we use Cassandra, it natively supports horizontal scaling utilizing consistent hashing
- If a new node is added to the cluster, data automatically gets rebalanced across all (virtual) nodes
- With this approach, no manual (re)sharding is required

Another scalability issue to consider is the hotspot issue - what if an ad is more popular and gets more attention than others?

- In the above example, aggregation service nodes can apply for extra resources via the resource manager
- The resource manager allocates more resources, so the original node isn't overloaded
- The original node splits the events into 3 groups and each of the aggregation nodes handles 100 events
- Result is written back to the original aggregation node

Alternative, more sophisticated ways to handle the hotspot problem:
- Global-Local Aggregation
- Split Distinct Aggregation

Fault Tolerance

Within the aggregation nodes, we are processing data in-memory. If a node goes down, the processed data is lost.

We can leverage consumer offsets in kafka to continue from where we left off once another node picks up the slack.
However, there is additional intermediary state we need to maintain, as we're aggregating the top N ads in M minutes.

We can make snapshots at a particular minute for the on-going aggregation:

If a node goes down, the new node can read the latest committed consumer offset, as well as the latest snapshot to continue the job:

Data monitoring and correctness

As the data we're aggregating is critical as it's used for billing, it is very important to have rigorous monitoring in place in order to ensure correctness.

Some metrics we might want to monitor:
- Latency: Timestamps of different events can be tracked in order to understand the e2e latency of the system
- Message queue size: If there is a sudden increase in queue size, we need to add more aggregation nodes. As Kafka is implemented via a distributed commit log, we need to keep track of records-lag metrics instead.
- System resources on aggregation nodes: CPU, disk, JVM, etc.

We also need to implement a reconciliation flow which is a batch job, running at the end of the day.
It calculates the aggregated results from the raw data and compares them against the actual data stored in the aggregation database:

Alternative design

In a generalist system design interview, you are not expected to know the internals of specialized software used in big data processing.

Explaining the thought process and discussing trade-offs is more important than knowing specific tools, which is why the chapter covers a generic solution.

An alternative design, which leverages off-the-shelf tooling, is to store ad click data in Hive with an ElasticSearch layer on top built for faster queries.

Aggregation is typically done in OLAP databases such as ClickHouse or Druid.

---

Step 4: Wrap up

Things we covered:
- Data model and API Design
- Using MapReduce to aggregate ad click events
- Scaling the message queue, aggregation service and database
- Mitigating the hotspot issue
- Monitoring the system continuously
- Using reconciliation to ensure correctness
- Fault tolerance

The ad click event aggregation is a typical big data processing system.

It would be easier to understand and design it if you have prior knowledge of related technologies:
- Apache Kafka
- Apache Spark
- Apache Flink

---

22. Hotel Reservation System/README (22. Hotel Reservation System/README.md)

Chapter 22: Hotel Reservation System

Introduction

In this chapter, we're designing a hotel reservation system, similar to Marriott International.

Applicable to other types of systems as well - Airbnb, flight reservation, movie ticket booking.

---

Step 1: Understand the Problem and Establish Design Scope

Before diving into designing the system, we should ask the interviewer questions to clarify the scope:
- C: What is the scale of the system?
- I: We're building a website for a hotel chain \w 5000 hotels and 1mil rooms
- C: Do customers pay when they make a reservation or when they arrive at the hotel?
- I: They pay in full when making reservations.
- C: Do customers book hotel rooms through the website only? Do we have to support other reservation options such as phone calls?
- I: They make bookings through the website or app only.
- C: Can customers cancel reservations?
- I: Yes
- C: Other things to consider?
- I: Yes, we allow overbooking by 10%. Hotel will sell more rooms than there actually are. Hotels do this in anticipation that clients will cancel bookings.
- C: Since not much time, we'll focus on - show hotel-related page, hotel-room details page, reserve a room, admin panel, support overbooking.
- I: Sounds good.
- I: One more thing - hotel prices change all the time. Assume a hotel room's price changes every day.
- C: OK.

Non-functional requirements

- Support high concurrency - there might be a lot of customers trying to book the same hotel during peak season.
- Moderate latency - it's ideal to have low latency when a user makes a reservation, but it's acceptable if the system takes a few seconds to process it.

Back-of-the-envelope estimation

- 5000 hotels and 1mil rooms in total
- Assume 70% of rooms are occupied and average stay duration is 3 days
- Estimated daily reservations - 1mil * 0.7 / 3 = ~240k reservations per day
- Reservations per second - 240k / 10^5 seconds in a day = ~3. Average reservation TPS is low.

Let's estimate the QPS. If we assume that there are three steps to reach the reservation page and there is a 10% conversion rate per page,
we can estimate that if there are 3 reservations, then there must be 30 views of reservation page and 300 views of hotel room detail page.

---

Step 2: Propose High-Level Design and Get Buy-In

We'll explore - API Design, Data model, high-level design.

API Design

This API Design focuses on the core endpoints (using RESTful practices), we'll need in order to support a hotel reservation system.

A fully-fledged system would require a more extensive API with support for searching for rooms based on lots of criteria, but we won't be focusing on that in this section.
Reason is that they aren't technically challenging, so they're out of scope.

Hotel-related API
- GET /v1/hotels/{id} - get detailed info about a hotel
- POST /v1/hotels - add a new hotel. Only available to ops
- PUT /v1/hotels/{id} - update hotel info. Only available to ops
- DELETE /v1/hotels/{id} - delete a hotel. API is only available to ops

Room-related API
- GET /v1/hotels/{id}/rooms/{id} - get detailed information about a room
- POST /v1/hotels/{id}/rooms - Add a room. Only available to ops
- PUT /v1/hotels/{id}/rooms/{id} - Update room info. Only available to ops
- DELETE /v1/hotels/{id}/rooms/{id} - Delete a room. Only available to ops

Reservation-related API
- GET /v1/reservations - get reservation history of current user
- GET /v1/reservations/{id} - get detailed info about a reservation
- POST /v1/reservations - make a new reservation
- DELETE /v1/reservations/{id} - cancel a reservation

Here's an example request to make a reservation:

text

{
  "startDate":"2021-04-28",
  "endDate":"2021-04-30",
  "hotelID":"245",
  "roomID":"U12354673389",
  "reservationID":"13422445"
}

Note that the reservationID is an idempotency key to avoid double booking. Details explained in concurrency section

Data model

Before we choose what database to use, let's consider our access patterns.

We need to support the following queries:
- View detailed info about a hotel
- Find available types of rooms given a date range
- Record a reservation
- Look up a reservation or past history of reservations

From our estimations, we know the scale of the system is not large, but we need to prepare for traffic surges.

Given this knowledge, we'll choose a relational database because:
- Relational DBs work well with read-heavy and less write-heavy systems.
- NoSQL databases are normally optimized for writes, but we know we won't have many as only a fraction of users who visit the site make a reservation.
- Relational DBs provide ACID guarantees. These are important for such a system as without them, we won't be able to prevent problems such as negative balance, double charge, etc.
- Relational DBs can easily model the data as the structure is very clear.

Here is our schema design:

Most fields are self-explanatory. Only field worth mentioning is the status field which represents the state machine of a given room:

This data model works well for a system like Airbnb, but not for hotels where users don't reserve a particular room but a room type.
They reserve a type of room and a room number is chosen at the point of reservation.

This shortcoming will be addressed in the Improved Data Model section.

High-level Design

We've chosen a microservice architecture for this design. It has gained great popularity in recent years:

- Users: book a hotel room on their phone or computer
- Admin: perform administrative functions such as refunding/cancelling a payment, etc
- CDN: caches static resources such as JS bundles, images, videos, etc
- Public API Gateway: fully-managed service which supports rate limiting, authentication, etc.
- Internal APIs: only visible to authorized personnel. Usually protected by a VPN.
- Hotel service: provides detailed information about hotels and rooms. Hotel and room data is static, so it can be cached aggressively.
- Rate service: provides room rates for different future dates. An interesting note about this domain is that prices depend on how full a hotel is at a given day.
- Reservation service: receives reservation requests and reserves hotel rooms. Also tracks room inventory as reservations are made/cancelled.
- Payment service: processes payments and updates reservation statuses on success.
- Hotel management service: available to authorized personnel only. Allows certain administrative functions for managing and viewing reservations, hotels, etc.

Inter-service communication can be facilitated via a RPC framework, such as gRPC.

---

Step 3: Design Deep Dive

Let's dive deeper into:
- Improved data model
- Concurrency issues
- Scalability
- Resolving data inconsistency in microservices

Improved data model

As mentioned in a previous section, we need to amend our API and schema to enable reserving a type of room vs. a particular one.

For the reservation API, we no longer reserve a roomID, but we reserve a roomTypeID:

text

POST /v1/reservations
{
  "startDate":"2021-04-28",
  "endDate":"2021-04-30",
  "hotelID":"245",
  "roomTypeID":"12354673389",
  "roomCount":"3",
  "reservationID":"13422445"
}

Here's the updated schema:

- room: contains information about a room
- room_type_rate: contains information about prices for a given room type
- reservation: records guest reservation data
- room_type_inventory: stores inventory data about hotel rooms.

Let's take a look at the room_type_inventory columns as that table is more interesting:
- hotel_id: id of hotel
- room_type_id: id of a room type
- date: a single date
- total_inventory: total number of rooms minus those that are temporarily taken off the inventory.
- total_reserved: total number of rooms booked for given (hotel_id, room_type_id, date)

There are alternative ways to design this table, but having one room per (hotel_id, room_type_id, date) enables easy
reservation management and easier queries.

The rows in the table are pre-populated using a daily CRON job.

Sample data:
| hotel_id | room_type_id | date | total_inventory | total_reserved |
|----------|--------------|------------|-----------------|----------------|
| 211 | 1001 | 2021-06-01 | 100 | 80 |
| 211 | 1001 | 2021-06-02 | 100 | 82 |
| 211 | 1001 | 2021-06-03 | 100 | 86 |
| 211 | 1001 | ... | ... | |
| 211 | 1001 | 2023-05-31 | 100 | 0 |
| 211 | 1002 | 2021-06-01 | 200 | 16 |
| 2210 | 101 | 2021-06-01 | 30 | 23 |
| 2210 | 101 | 2021-06-02 | 30 | 25 |

Sample SQL query to check the availability of a type of room:

text

SELECT date, total_inventory, total_reserved
FROM room_type_inventory
WHERE room_type_id = ${roomTypeId} AND hotel_id = ${hotelId}
AND date between ${startDate} and ${endDate}

How to check availability for a specified number of rooms using that data (note that we support overbooking):

text

if (total_reserved + ${numberOfRoomsToReserve}) <= 110% * total_inventory

Now let's do some estimation about the storage volume.
- We have 5000 hotels.
- Each hotel has 20 types of rooms.
- 5000 20 2 (years) * 365 (days) = 73mil rows

73 million rows is not a lot of data and a single database server can handle it.
It makes sense, however, to setup read replication (potentially across different zones) to enable high availability.

Follow-up question - if reservation data is too large for a single database, what would you do?
- Store only current and future reservation data. Reservation history can be moved to cold storage.
- Database sharding - we can shard our data by hash(hotel_id) % servers_cnt as we always select the hotel_id in our queries.

Concurrency issues

Another important problem to address is double booking.

There are two issues to address:
- Same user clicks on "book" twice
- Multiple users try to book a room at the same time

Here's a visualization of the first problem:

There are two approaches to solving this problem:
- Client-side handling - front-end can disable the book button once clicked. If a user disabled javascript, however, they won't see the button becoming grayed out.
- Idemptent API - Add an idempotency key to the API, which enables a user to execute an action once, regardless of how many times the endpoint is invoked:

Here's how this flow works:
- A reservation order is generated once you're in the process of filling in your details and making a booking. The reservation order is generated using a globally unique identifier.
- Submit reservation 1 using the reservation_id generated in the previous step.
- If "complete booking" is clicked a second time, the same reservation_id is sent and the backend detects that this is a duplicate reservation.
- The duplication is avoided by making the reservation_id column have a unique constraint, preventing multiple records with that id being stored in the DB.

What if there are multiple users making the same reservation?

- Let's assume the transaction isolation level is not serializable
- User 1 and 2 attempt to book the same room at the same time.
- Transaction 1 checks if there are enough rooms - there are
- Transaction 2 check if there are enough rooms - there are
- Transaction 2 reserves the room and updates the inventory
- Transaction 1 also reserves the room as it still sees there are 99 total_reserved rooms out of 100.
- Both transactions successfully commit the changes

This problem can be solved using some form of locking mechanism:
- Pessimistic locking
- Optimistic locking
- Database constraints

Here's the SQL we use to reserve a room:

sql

step 1: check room inventory

SELECT date, total_inventory, total_reserved
FROM room_type_inventory
WHERE room_type_id = ${roomTypeId} AND hotel_id = ${hotelId}
AND date between ${startDate} and ${endDate}
For every entry returned from step 1

if((total_reserved + ${numberOfRoomsToReserve}) > 110% * total_inventory) {
  Rollback
}
step 2: reserve rooms

UPDATE room_type_inventory
SET total_reserved = total_reserved + ${numberOfRoomsToReserve}
WHERE room_type_id = ${roomTypeId}
AND date between ${startDate} and ${endDate}Commit

#### Option 1: Pessimistic locking
Pessimistic locking prevents simultaneous updates by putting a lock on a record while it's being updated.

This can be done in MySQL by using the SELECT... FOR UPDATE query, which locks the rows selected by the query until the transaction is committed.

Pros:
- Prevents applications from updating data that is being changed
- Easy to implement and avoids conflict by serializing updates. Useful when there is heavy data contention.

Cons:
- Deadlocks may occur when multiple resources are locked.
- This approach is not scalable - if transaction is locked for too long, this has impact on all other transactions trying to access the resource.
- The impact is severe when the query selects a lot of resources and the transaction is long-lived.

The author doesn't recommend this approach due to its scalability issues.

#### Option 2: Optimistic locking
Optimistic locking allows multiple users to attempt to update a record at the same time.

There are two common ways to implement it - version numbers and timestamps. Version numbers are recommended as server clocks can be inaccurate.

- A new version column is added to the database table
- Before a user modifies a database row, the version number is read
- When the user updates the row, the version number is increased by 1 and written back to the database
- Database validation prevents the insert if the new version number doesn't exceed the previous one

Optimistic locking is usually faster than pessimistic locking as we're not locking the database.
Its performance tends to degrade when concurrency is high, however, as that leads to a lot of rollbacks.

Pros:
- It prevents applications from editing stale data
- We don't need to acquire a lock in the database
- Preferred option when data contention is low, ie rarely are there update conflicts

Cons:
- Performance is poor when data contention is high

Optimistic locking is a good option for our system as reservation QPS is not extremely high.

#### Option 3: Database constraints
This approach is very similar to optimistic locking, but the guardrails are implemented using a database constraint:

text

CONSTRAINT check_room_count CHECK((total_inventory - total_reserved >= 0))

Pros:
- Easy to implement
- Works well when data contention is small

Cons:
- Similar to optimistic locking, performs poorly when data contention is high
- Database constraints cannot be easily version-controlled like application code
- Not all databases support constraints

This is another good option for a hotel reservation system due to its ease of implementation.

Scalability

Usually, the load of a hotel reservation system is not high.

However, the interviewer might ask you how you'd handle a situation where the system gets adopted for a larger, popular travel site such as booking.com
In that case, QPS can be 1000 times larger.

When there is such a situation, it is important to understand where our bottlenecks are. All the services are stateless, so they can be easily scaled via replication.

The database, however, is stateful and it's not as obvious how it can get scaled.

One way to scale it is by implementing database sharding - we can split the data across multiple databases, where each of them contain a portion of the data.

We can shard based on hotel_id as all queries filter based on it.
Assuming, QPS is 30,000, after sharding the database in 16 shards, each shard handles 1875 QPS, which is within a single MySQL cluster's load capacity.

We can also utilize caching for room inventory and reservations via Redis. We can set TTL so that old data can expire for days which are past.

The way we store an inventory is based on the hotel_id, room_type_id and date:

text

key: hotelID_roomTypeID_{date}
value: the number of available rooms for the given hotel ID, room type ID and date.

Data consistency happens async and is managed by using a CDC streaming mechanism - database changes are read and applied to a separate system.
Debezium is a popular option for synchronizing database changes with Redis.

Using such a mechanism, there is a possibility that the cache and database are inconsistent for some time.
This is fine in our case because the database will prevent us from making an invalid reservation.

This will cause some issue on the UI as a user would have to refresh the page to see that "there are no more rooms left",
but that is something which can happen regardless of this issue if eg a person hesitates a lot before making a reservation.

Caching pros:
- Reduced database load
- High performance, as Redis manages data in-memory

Caching cons:
- Maintaining data consistency between cache and DB is hard. We need to consider how the inconsistency impacts user experience.

Data consistency among services

A monolithic application enables us to use a shared relational database for ensuring data consistency.

In our microservice design, we chose a hybrid approach where some services are separate,
but the reservation and inventory APIs are handled by the same servicefor the reservation and inventory APIs.

This is done because we want to leverage the relational database's ACID guarantees to ensure consistency.

However, the interviewer might challenge this approach as it's not a pure microservice architecture, where each service has a dedicated database:

This can lead to consistency issues. In a monolithic server, we can leverage a relational DBs transaction capabilities to implement atomic operations:

It's more challenging, however, to guarantee this atomicity when the operation spans across multiple services:

There are some well-known techniques to handle these data inconsistencies:
- Two-phase commit: a database protocol which guarantees atomic transaction commit across multiple nodes.
It's not performant, though, since a single node lag leads to all nodes blocking the operation.
- Saga: a sequence of local transactions, where compensating transactions are triggered if any of the steps in a workflow fail. This is an eventually consistent approach.

It's worth noting that addressing data inconsistencies across microservices is a challenging problem, which raise the system complexity.
It is good to consider whether the cost is worth it, given our more pragmatic approach of encapsulating dependent operations within the same relational database.

---

Step 4: Wrap Up

We presented a design for a hotel reservation system.

These are the steps we went through:
- Gathering requirements and doing back-of-the-envelope calculations to understand the system's scale
- We presented the API Design, Data Model and system architecture in the high-level design
- In the deep dive, we explored alternative database schema designs as requirements changed
- We discussed race conditions and proposed solutions - pessimistic/optimistic locking, database constraints
- Ways to scale the system via database sharding and caching
- Finally we addressed how to handle data consistency issues across multiple microservices

---

23. Distributed Email Service/README (23. Distributed Email Service/README.md)

Chapter 23: Distributed Email Service

Introduction

We'll design a distributed email service, similar to Gmail in this chapter.

In 2020, Gmail had 1.8bil active users, while Outlook had 400mil users worldwide.

---

Step 1: Understand the Problem and Establish Design Scope

- C: How many users use the system?
- I: 1bil users
- C: I think following features are important - auth, send/receive email, fetch email, filter emails, search email, anti-spam protection.
- I: Good list. Don't worry about auth for now.
- C: How do users connect \w email servers?
- I: Typically, email clients connect via SMTP, POP, IMAP, but we'll use HTTP for this problem.
- C: Can emails have attachments?
- I: Yes

Non-functional requirements

- Reliability - we shouldn't lose data
- Availability - We should use replication to prevent single points of failure. We should also tolerate partial system failures.
- Scalability - As userbase grows, our system should be able to handle them.
- Flexibility and extensibility - system should be flexible and easy to extend with new features. One of the reasons we chose HTTP over SMTP/other mail protocols.

Back-of-the-envelope estimation

- 1bil users
- Assuming one person sends 10 emails per day -> 100k emails per second.
- Assuming one person receives 40 emails per day and each email on average has 50kb metadata -> 730pb storage per year.
- Assuming 20% of emails have storage attachments and average size is 500kb -> 1,460pb per year.

---

Step 2: Propose High-Level Design and Get Buy-In

Email knowledge 101

There are various protocols used for sending and receiving emails:
- SMTP - standard protocol for sending emails from one server to another.
- POP - standard protocol for receiving and downloading emails from a remote mail server to a local client. Once retrieved, emails are deleted from remote server.
- IMAP - similar to POP, it is used for receiving and downloading emails from a remote server, but it keeps the emails on the server-side.
- HTTPS - not technically an email protocol, but it can be used for web-based email clients.

Apart from the mailing protocol, there are some DNS records we need to configure for our email server - the MX records:

Email attachments are sent base64-encoded and there is usually a size limit of 25mb on most mail services.
This is configurable and varies from individual to corporate accounts.

Traditional mail servers

Traditional mail servers work well when there are a limited number of users, connected to a single server.

- Alice logs into her Outlook email and presses "send". Email is sent to Outlook mail server. Communication is via SMTP.
- Outlook server queries DNS to find MX record for gmail.com and transfers the email to their servers. Communication is via SMTP.
- Bob fetches emails from his gmail server via IMAP/POP.

In traditional mail servers, emails were stored on the local file system. Every email was a separate file.

As the scale grew, disk I/O became a bottleneck. Also, it doesn't satisfy our high availability and reliability requirements.
Disks can be damaged and server can go down.

Distributed mail servers

Distributed mail servers are designed to support modern use-cases and solve modern scalability issues.

These servers can still support IMAP/POP for native email clients and SMTP for mail exchange across servers.

But for rich web-based mail clients, a RESTful API over HTTP is typically used.

Example APIs:
- POST /v1/messages - sends a message to recipients in To, Cc, Bcc headers.
- GET /v1/folders - returns all folders of an email account

Example response:

text

[{id: string        Unique folder identifier.
  name: string      Name of the folder.
                    According to RFC6154 [9], the default folders can be one of
                    the following: All, Archive, Drafts, Flagged, Junk, Sent,
                    and Trash.
  user_id: string   Reference to the account owner
}]

- GET /v1/folders/{:folder_id}/messages - returns all messages under a folder \w pagination
- GET /v1/messages/{:message_id} - get all information about a particular message

Example response:

text

{
  user_id: string                      // Reference to the account owner.
  from: {name: string, email: string}  // <name, email> pair of the sender.
  to: [{name: string, email: string}]  // A list of <name, email> paris
  subject: string                      // Subject of an email
  body: string                         //  Message body
  is_read: boolean                     //  Indicate if a message is read or not.
}

Here's the high-level design of the distributed mail server:

- Webmail - users use web browsers to send/receive emails
- Web servers - public-facing request/response services used to manage login, signup, user profile, etc.
- Real-time servers - Used for pushing new email updates to clients in real-time. We use websockets for real-time communication but fallback to long-polling for older browsers that don't support them.
- Metadata db - stores email metadata such as subject, body, from, to, etc.
- Attachment store - Object store (eg Amazon S3), suitable for storing large files.
- Distributed cache - We can cache recent emails in Redis to improve UX.
- Search store - distributed document store, used for supporting full-text searches.

Here's what the email sending flow looks like:

- User writes an email and presses "send". Email is sent to load balancer.
- Load balancer rate limits excessive mail sends and routes to one of the web servers.
- Web servers do basic email validation (eg email size) and short-circuits outbound flow if domain is same as sender. But does spam check first.
- If basic validation passes, email is sent to message queue (attachment is referenced from object store)
- If basic validation fails, email is sent to error queue
- SMTP outgoing workers pull messages from outgoing queue, do spam/virus checks and route to destination mail server.
- Email is stored in the "Sent Emails" folder

We need to also monitor size of outgoing message queue. Growing too large might indicate a problem:
- Recipient's mail server is unavailable. We can retry sending the email at a later time using exponential backoff.
- Not enough consumers to handle the load, we might have to scale the consumers.

Here's the email receiving flow:

- Incoming emails arrive at the SMTP load balancer. Mails are distributed to SMTP servers, where mail acceptance policy is done (eg invalid emails are directly discarded).
- If attachment of email is too large, we can put it in object store (s3).
- Mail processing workers do preliminary checks, after which mails are forwarded to storage, cache, object store and real-time servers.
- Offline users get their new emails once they come back online via HTTP API.

---

Step 3: Design Deep Dive

Let's now go deeper into some of the components.

Metadata database

Here are some of the characteristics of email metadata:
- headers are usually small and frequently accessed
- Body size ranges from small to big, but is typically read once
- Most mail operations are isolated to a single user - eg fetching email, marking as read, searching.
- Data recency impacts data usage. Users typically read only recent emails
- Data has high-reliability requirements. Data loss is unacceptable.

At gmail/outlook scale, the database is typically custom made to reduce input/output operations per second (IOPS).

Let's consider what database options we have:
- Relational database - we can build indexes for headers and body, but these DBs are typically optimized for small chunks of data.
- Distributed object store - this can be a good option for backup storage, but can't efficiently support searching/marking as read/etc.
- NoSQL - Google BigTable is used by gmail, but it's not open-sourced.

Based on above analysis, very few existing solutions seems to fit our needs perfectly.
In an interview setting, it's infeasible to design a new distributed database solution, but important to mention characteristics:
- Single column can be a single-digit MB
- Strong data consistency
- Designed to reduce disk I/O
- Highly available and fault tolerant
- Should be easy to create incremental backups

In order to partition the data, we can use the user_id as a partition key, so that one user's data is stored on a single shard.
This prohibits us from sharing an email with multiple users, but this is not a requirement for this interview.

Let's define the tables:
- Primary key consists of partition key (data distribution) and clustering key (sorting data)
- Queries we need to support - get all folders for a user, display all emails for a folder, create/get/delete an email, fetch read/unread email, get conversation threads (bonus)

Legend for tables to follow:

Here is the folders table:

emails table:

- email_id is timeuuid which allows sorting based on timestamp when email was created

Attachments are stored in a separate table, identified by filename:

Supporting fetchin read/unread emails is easy in a traditional relational database, but not in Cassandra, since filtering on non-partition/clustering key is prohibited.
One workaround to fetch all emails in a folder and filter in-memory, but that doesn't work well for a big-enough application.

What we can do is denormalize the emails table into read/unread emails tables:

In order to support conversation threads, we can include some headers, which mail clients interpret and use to reconstruct a conversation thread:

text

{
  "headers" {
     "Message-Id": "<[email protected]>",
     "In-Reply-To": "<[email protected]>",
     "References": ["<[email protected]>"]
  }
}

Finally, we'll trade availability for consistency for our distributed database, since it is a hard requirement for this problem.

Hence, in the event of a failover or network parititon, sync/update actions will be briefly unavailable to impacted users.

Email deliverability

It is easy to setup a server to send emails, but getting the email to a receiver's inbox is hard, due to spam-protection algorithms.

If we just setup a new mail server and start sending mails through it, our emails will probably end up in the spam folder.

Here's what we can do to prevent that:
- Dedicated IPs - use dedicated IPs for sending emails, otherwise, recipient servers will not trust you.
- Classify emails - avoid sending marketing emails from the same servers to prevent more important email to be classified as spam
- Warm up your IP address slowly to build a good reputation with big email providers. It takes 2 to 6 weeks to warm up a new IP
- Ban spammers quickly to not deteriorate your reputation
- Feedback processing - setup a feedback loop with ISPs to keep track of complaint rate and ban spam accounts quickly.
- Email authentication - use common techniques to combat phishing such as Sender Policy Framework, DomainKeys Identified Mail, etc.

You don't need to remember all of this. Just know that building a good mail server requires a lot of domain knowledge.

Search

Searching includes doing a full-text search based on email contents or more advanced queries based on from, to, subject, unread, etc filters.

One characteristic of email search is that it is local to the user and it has more writes than reads, because we need to re-index it on each operation, but users rarely use the search tab.

Let's compare google search with email search:

| | Scope | Sorting | Accuracy |
|---------------|----------------------|---------------------------------------|---------------------------------------------------|
| Google search | The whole internet | Sort by relevance | Indexing takes some time, so not instant results. |
| Email search | User's own email box | Sort by attributes eg time, date, etc | Indexing should be quick and results accurate. |

To achieve this search functionality, one option is to use an Elasticsearch cluster. We can use user_id as the partition key to group data under the same node:

Mutating operations are async via Kafka in order to decouple services from the reindexing flow.
Actually searching for data happens synchronously.

Elasticsearch is one of the most popular search-engine databases and supports full-text search for emails very well.

Alternatively, we can attempt to develop our own custom search solution to meet our specific requirements.

Designing such a system is out of scope. One of the core challenges when building it is to optimize it for write-heavy workloads.

To achieve that, we can use Log-Structured Merge-Trees (LSM) to structure the index data on disk. Write path is optimized for sequential writes only.
This technique is used in Cassandra, BigTable and RocksDB.

Its core idea is to store data in-memory until a predefined threshold is reached, after which it is merged in the next layer (disk):

Main trade-offs between the two approaches:
- Elasticsearch scales to some extent, whereas a custom search engine can be fine-tuned for the email use-case, allowing it to scale further.
- Elasticsearch is a separate service we need to maintain, alongside the metadata store. A custom solution can be the datastore itself.
- Elasticsearch is an off-the-shelf solution, whereas the custom search engine would require significant engineering effort to build.

Scalability and availability

Since individual user operations don't collide with other users, most components can be independently scaled.

To ensure high availability, we can also use a multi-DC setup with leader-folower failover in case of failures:

---

Step 4: Wrap Up

Additional talking points:
- Fault tolerance - Many parts of the system could fail. It is worthwhile how we'd handle node failures.
- Compliance - PII needs to be stored in a reasonable way, given Europe's GDPR laws.
- Security - email encryption, phishing protection, safe browsing, etc.
- Optimizations - eg preventing duplication of the same attachments, sent multiple times by different users.

---

24. S3 Like Object Storage/README (24. S3-like Object Storage/README.md)

Chapter 24: S3-like Object Storage

Introduction

In this chapter, we'll be designing an object storage service, similar to Amazon S3.

Storage systems fall into three broad categories:
- Block storage
- File storage
- Object storage

Block storage are devices, which came out in 1960s. HDDs and SSDs are such examples.
These devices are typically physically attached to a server, although they can also be network-attached via high-speed network protocols.
Servers can format the raw blocks and use them as a file system or it can hand control of them to servers directly.

File storage is built on top of block storage. It provides a higher level of abstraction, making it easier to manage folders and files.

Object storage sacrifices performance for high durability, vast scale and low cost.
It targets "cold" data and is mainly used for archival and backup.
There is no hierarchical directory structure, all data is stored as objects in a flat structure.
It is relatively slow compared to other storage types. Most cloud providers have an object storage offering - Amazon S3, Google GCS, etc.

| | Block Storage | File Storage | Object Storage |
|-----------------|----------------------------------|-----------------------------------------|--------------------------------|
| Mutable Content | Y | Y | N (has object versioning） |
| Cost | High | Medium to high | Low |
| Performance | Medium to high, very high | Medium to high | Low to medium |
| Consistency | Strong consistency | Strong consistency | Strong consistency [5] |
| Data access | SAS/iSCSI/FC | Standard file access, CIFS/SMB, and NFS | RESTful API |
| Scalability | Medium scalability | High scalability | Vast scalability |
| Good for | Virtual machines (VM), databases | General-purpose file system access | Binary data, unstructured data |

Some terminology, related to object storage:
- Bucket - logical container for objects. Name is globally unique.
- Object - An individual piece of data, stored in a bucket. Contains object data and metadata.
- Versioning - A feature keeping multiple variants of an object in the same bucket.
- Uniform Resource Identifier (URI) - each resource is uniquely identified by a URI.
- Service-level Agreement (SLA) - contract between service provider and client.

Amazon S3 Standard-Infrequent Access storage class SLAs:
- Durability of 99.999999999% across multiple Availability Zones
- Data is resilient in the event of entire Availability Zone being destroyed
- Designed for 99.9% availability

---

Step 1: Understand the Problem and Establish Design Scope

- C: Which features should be included?
- I: Bucket creation, Object upload/download, versioning, Listing objects in a bucket
- C: What is the typical data size?
- I: We need to store both massive objects and small objects efficiently
- C: How much data do we store in a year?
- I: 100 petabytes
- C: Can we assume 6 nines of data durbility (99.9999%) and service availability of 4 nines (99.99%)?
- I: Yes, sounds reasonable

Non-functional requirements

- 100 PB of data
- 6 nines of data durability
- 4 nines of service availability
- Storage efficiency. Reduce storage cost while maintaining high reliability and performance

Back-of-the-envelope estimation

Object storage is likely to have bottlenecks in disk capacity or IO per second (IOPS).

Assumptions:
- we have 20% small (less than 1mb), 60% mid-size (1-64mb) and 20% large objects (greater than 64mb),
- One hard disk (SATA, 7200rpm) is capable of doing 100-150 random seeks per second (100-150 IOPS)

Given the assumptions, we can estimate the total number of objects the system can persist.
- Let's use median size per object type to simplify calculation - 0.5mb for small, 32mb for medium, 200mb for large.
- Given 100PB of storage (10^11 MB) and 40% of storage usage results in 0.68bil objects
- If we assume metadata is 1kb, then we need 0.68tb space to store metadata info

---

Step 2: Propose High-Level Design and Get Buy-In

Let's explore some interesting properties of object storage before diving into the design:
- Object immutability - objects in object storage are immutable (not the case in other storage systems). We may delete them or replace them, but no update.
- Key-value store - an object URI is its key and we can get its contents by making an HTTP call
- Write once, read many times - data access pattern is writing once and reading many times. According to some Linkedin research, 95% of operations are reads
- Support both small and large objects

Design philosophy of object storage is similar to UNIX - when we save a file, it creates the filename in a data structure, called inode and file data is stored in different disk locations.
The inode contains a list of file block pointers, which point to different locations on disk.

When accessing a file, we first fetch its metadata from the inode, prior to fetching the file contents.

Object storage works similarly - metadata store is used for file information, but contents are stored on disk:

By separating metadata from file contents, we can scale the different stores independently:

High-level design

- Load balancer - distributes API requests across service replicas
- API service - Stateless server, orchestrating calls to metadata and object store, as well as IAM service.
- Identity and access management (IAM) - central place for auth, authz, access control.
- Data store - stores and retrieves actual data. Operations are based on object ID (UUID).
- Metadata store - stores object metadata

Uploading an object

- Create a bucket named "bucket-to-share" via HTTP PUT request
- API service calls IAM to ensure user is authorized and has write permissions
- API service calls metadata store to create a bucket entry. Once created, success response is returned.
- After bucket is created, HTTP PUT is sent to create an object named "script.txt"
- API service verifies user identity and ensures user has write permissions
- Once validation passes, object payload is sent via HTTP PUT to the data store. Data store persists it and returns a UUID.
- API service calls metadata store to create a new entry with object_id, bucket_id and bucket_name, among other metadata.

Example object upload request:

text

PUT /bucket-to-share/script.txt HTTP/1.1 Host: foo.s3example.org Date: Sun, 12 Sept 2021 17:51:00 GMT Authorization: authorization string Content-Type: text/plain Content-Length: 4567 x-amz-meta-author: Alex

[4567 bytes of object data]

Downloading an object

Buckets have no directory hierarchy, buy we can create a logical hierarchy by concatenating bucket name and object name to simulate a folder structure.

Example GET request for fetching an object:

text

GET /bucket-to-share/script.txt HTTP/1.1
Host: foo.s3example.org
Date: Sun, 12 Sept 2021 18:30:01 GMT
Authorization: authorization string

- Client sends an HTTP GET request to the load balancer, ie GET /bucket-to-share/script.txt
- API service queries IAM to verify the user has correct permissions to read the bucket
- Once validated, UUID of object is retrieved from metadata store
- Object payload is retrieved from data store based on UUID and returned to the client

---

// sprint 1

Step 3: Design Deep Dive

Data store

Here's how the API service interacts with the data store:

The data store's main components:

The data routing service provides a RESTful or gRPC API to access the data node cluster.
It is a stateless service, which scales by adding more servers.

It's main responsibilities are:
- querying the placement service to get the best data node to store data
- reading data from data nodes and returning it to the API service
- Writing data to data nodes

The placement service determines which data nodes should store an object.
It maintains a virtual cluster map, which determines the physical topology of a cluster.

The service also sends heartbeats to all data nodes to determine if they should be removed from the virtual cluster.

Since this is a critical service, it is recommended to maintain a cluster of 5 or 7 replicas, synchronized via Paxos or Raft consensus algorithms.
Eg a 7 node cluster can tolerate 3 nodes failing.

Data nodes store the actual object data.
Reliability and durability is ensured by replicating data to multiple data nodes.

Each data node has a daemon running, which sends heartbeats to the placement service.

The heartbeat includes:
- How many disk drives (HDD or SSD) does the data node manage?
- How much data is stored on each drive?

#### Data persistence flow

- API service forwards the object data to data store
- Data routing service sends the data to the primary data node
- Primary data node saves the data locally and replicates it to two secondary data nodes. Response is sent after successful replication.
- The UUID of the object is returned to the API service.

Caveats:
- Given an object UUID, it's replication group is deterministically chosen by using consistent hashing
- In step 4, the primary data node replicates the object data before returning a response. This favors strong consistency over higher latency.

#### How data is organized

One simple approach to managing data is to store each object in a separate file.

This works, but is not performant with many small files in a file system:
- Data blocks on HDD are wasted, because every file uses the whole block size. Typical block size is 4kb.
- Many files means many inodes. Operating systems don't deal well with too many inodes and there is also a max inode limit.

These issues can be addressed by merging many small files into bigger ones via a write-ahead log (WAL). Once the file reaches its capacity (typically a few GB), a new file is created:

The downside of this approach is that write access to the file needs to be serialized. Multiple cores accessing the same file must wait for each other.
To fix this, we can confine files to specific cores to avoid lock contention.

#### Object lookup

To support storing multiple objects in the same file, we need to maintain a table, which tells the data node:
- object_id
- filename where object is stored
- file_offset where object starts
- object_size

We can deploy this table in a file-based db like RocksDB or a traditional relational database.
Since the access pattern is low write+high read, a relational database works better.

How should we deploy it?
We could deploy the db and scale it separately in a cluster, accessed by all data nodes.

Downsides:
- we'd need to aggressively scale the cluster to serve all requests
- there's additional network latency between data node and db cluster

An alternative is to take advantage of the fact that data nodes are only interested to data related to them,
so we can deploy the relational db within the data node itself.

SQLite is a good option as it's a lightweight file-based relational database.

#### Updated data persistence flow

- API Service sends a request to save a new object
- Data node service appends the new object at the end of a file, named "/data/c"
- A new record for the object is inserted into the object mapping table

#### Durability

Data durability is an important requirement in our design. In order to achieve 6 nines of durability, every failure case needs to be properly examined.

First problem to address is hardware failures. We can achieve that by replicating data nodes to minimize probability of failure.
But in addition to that, we also ought to replicate across different failure domains (cross-rack, cross-dc, separate networks, etc).
A critical event can cause multiple hardware failures within the same domain:

Assuming annual failure rate of a typical HDD is 0.81%, making three copies gives us 6 nines of durability.

Replicating the data nodes like that grants us the durability we want, but we could also leverage erasure coding to reduce storage costs.

Erasure coding enables us to use parity bits, which allow us to reconstruct lost bits in the event of a failure:

Imagine those bits are data nodes. If two of them go down, they can be recovered using the remaining four ones.

There are different erasure coding schemes. In our case, we could use 8+4 erasure coding, split across different failure domains to maximize reliability:

Erasure coding enables us to achieve a much lower storage cost (50% improvement) at the expense of access speed due to the data routing service having to collect data from multiple locations:

Other caveats:
- Replication requires 200% storage overhead (in case of 3 replicas) vs. 50% via erasure coding
- Erasure coding gives us 11 nines of durability vs 6 nines via replication
- Erasure coding requires more computation to calculate and store parities

In sum, replication is more useful for latency-sensitive applications, whereas erasure coding is attractive for storage cost efficiency and durability.
Erasure coding is also much harder to implement.

#### Correctness verification

If a disk fails entirely, then the failure is easy to detect. This is less straightforward in the event part of the disk memory gets corrupted.

To detect this, we can use checksums - a hash of the file contents, which can be used to verify the file's integrity.

In our case, we'll store checksums for each file and each object:

In the case of erasure coding (8+4), we'll need to fetch each of the 8 pieces of data separately and verify each of their checksums.

// sprint 2

Metadata data model

Table schemas:

Queries we need to support:
- Find an object ID by name
- Insert/delete object based on name
- List objects in a bucket sharing the same prefix

There is usually a limit on the number of buckets a user can create, hence, the size of the buckets table is small and can fit into a single db server.
But we still need to scale the server for read throughput.

The object table will probably not fit into a single database server, though. Hence, we can scale the table via sharding:
- Sharding by bucket_id will lead to hotspot issues as a bucket can have billions of objects
- Sharding by bucket_id makes the load more evenly distributed, but our queries will be slow
- We choose sharding by hash(bucket_name, object_name) since most queries are based on the object/bucket name.

Even with this sharding scheme, though, listing objects in a bucket will be slow.

Listing objects in a bucket

In a single database, listing an object based on its prefix (looks like a directory) works like this:

text

SELECT * FROM object WHERE bucket_id = "123" AND object_name LIKE abc/%

This is challenging to fulfill when the database is sharded. To achieve it, we can run the query on every shard and aggregate the results in-memory.
This makes pagination challenging though, since different shards contain a different result size and we need to maintain separate limit/offset for each.

We can leverage the fact that typically object stores are not optimized for listing objects, so we can sacrifice listing performance.
We can also create a denormalized table for listing objects, sharded by bucket ID.
That would make our listing query sufficiently fast as it's isolated to a single database instance.

Object versioning

Versioning works by having another object_version column which is of type TIMEUUID, enabling us to sort records based on it.

Each new version produces a new object_id:

Deleting an object creates a new version with a special object_id indicating that the object was deleted. Queries for it return 404:

Optimizing uploads of large files

Uploading large files can be optimized by using multipart uploads - splitting a big file into several chunks, uploaded independently:

- Client calls service to initiate a multipart upload
- Data store returns an upload ID which uniquely identifies the upload
- Client splits the large file into several chunks, uploaded independently using the upload id
- When a chunk is uploaded, the data store returns an etag, which is a md5 checksum, identifying that upload chunk
- After all parts are uploaded, client sends a complete multipart upload request, which includes upload_id, part numbers and all etags
- Data store reassembles the object from its parts. The process can take a few minutes. After that, success response is returned to the client.

Old parts, which are no longer useful can be removed at this point. We can introduce a garbage collector to deal with it.

Garbage collection

Garbage collection is the process of reclaiming storage space, which is no longer used. There are a few ways data becomes garbage:
- lazy object deletion - object is marked as deleted without actually getting deleted
- orphan data - eg an upload failed mid-flight and old parts need to be deleted
- corrupted data - data which failed checksum verification

The garbage collector is also responsible for reclaiming unused space in replicas.
With replication, data is deleted from both primaries and replicas. With erasure coding (8+4), data is deleted from all 12 nodes.

To facilitate the deletion, we'll use a process called compaction:
- Garbage collector copies objects which are not deleted from "data/b" to "data/d"
- object_mapping table is updated once copying is complete using a database transaction
- To avoid making too many small files, compaction is done on files which grow beyond a certain threshold

---

Step 4: Wrap Up

Things we covered:
- Designing an S3-like object storage
- Comparing differences between object, block and file storages
- Covered uploading, downloading, listing, versioning of objects in a bucket
- Deep dived in the design - data store and metadata store, replication and erasure coding, multipart uploads, sharding

---

25. Real Time Gaming Leaderboard/README (25. Real-time Gaming Leaderboard/README.md)

Chapter 25: Real-time Gaming Leaderboard

Introduction

We are going to design a leaderboard for an online mobile game:

---

Step 1: Understand the Problem and Establish Design Scope

- C: How is the score calculated for the leaderboard?
- I: User gets a point whenever they win a match.
- C: Are all players included in the leaderboard?
- I: Yes
- C: Is there a time segment, associated with the leaderboard?
- I: Each month, a new tournament starts which starts a new leaderboard.
- C: Can we assume we only care about top 10 users?
- I: We want to display top 10 users, along with position of specific user. If time permits, we can discuss showing users around particular user in the leaderboard.
- C: How many users are in a tournament?
- I: 5mil DAU and 25mil MAU
- C: How many matches are played on average during a tournament?
- I: Each player plays 10 matches per day on average
- C: How do we determine the rank if two players have the same score?
- I: Their rank is the same in that case. If time permits, we can discuss breaking ties.
- C: Does the leaderboard need to be real-time?
- I: Yes, we want to present real-time results or as close as possible to real-time. It is not okay to present batched result history.

Functional requirements

- Display top 10 players on leaderboard
- Show a user's specific rank
- Display users which are four places above and below given user (bonus)

Non-functional requirements

- Real-time updates on scores
- Score update is reflected on the leaderboard in real-time
- General scalability, availability, reliability

Back-of-the-envelope estimation

With 50mil DAU, if the game has an even distribution of players during a 24h period, we'd have an average of 50 users per second.
However, since distribution is typically uneven, we can estimate that the peak online users would be 250 users per second.

QPS for users scoring a point - given 10 games per day on average, 50 users/s * 10 = 500 QPS. Peak QPS = 2500.

QPS for fetching the top 10 leaderboard - assuming users open that once a day on average, QPS is 50.

---

Step 2: Propose High-Level Design and Get Buy-In

API Design

The first API we need is one to update a user's score:

text

POST /v1/scores

This API takes two params - user_id and points scored for winning a game.

This API should only be accessible to game servers, not end clients.

Next one is for getting the top 10 players of the leaderboard:

text

GET /v1/scores

Example response:

text

{
  "data": [
    {
      "user_id": "user_id1",
      "user_name": "alice",
      "rank": 1,
      "score": 12543
    },
    {
      "user_id": "user_id2",
      "user_name": "bob",
      "rank": 2,
      "score": 11500
    }
  ],
  ...
  "total": 10
}

You can also get the score of a particular user:

text

GET /v1/scores/{:user_id}

Example response:

text

{
    "user_info": {
        "user_id": "user5",
        "score": 1000,
        "rank": 6,
    }
}

High-level architecture

- When a player wins a game, client sends a request to the game service
- Game service validates if win is valid and calls the leaderboard service to update the player's score
- Leaderboard service updates the user's score in the leaderboard store
- Player makes a call to leaderboard service to fetch leaderboard data, eg top 10 players and given player's rank

An alternative design which was considered is the client updating their score directly within the leaderboard service:

This option is not secure as it's susceptible to man-in-the-middle attacks. Players can put a proxy and change their score as they please.

One additional caveat is that for games, where the game logic is managed by the server, cliets don't need to call the server explicitly to record their win.
Servers do it automatically for them based on the game logic.

One additional consideration is whether we should put a message queue between the game server and the leaderboard service. This would be useful if other services are interested in game results, but that is not an explicit requirement in the interview so far, hence it's not included in the design:

Data models

Let's discuss the options we have for storing leaderboard data - relational DBs, Redis, NoSQL.

The NoSQL solution is discussed in the deep dive section.

#### Relational database solution

If the scale doesn't matter and we don't have that many users, a relational DB serves our quite well.

We can start from a simple leaderboard table, one for each month (personal note - this doesn't make sense. You can just add a month column and avoid the headache of maintaining new tables each month):

There is additional data to include in there, but that is irrelevant to the queries we'd run, so it's omitted.

What happens when a user wins a point?

If a user doesn't exist in the table yet, we need to insert them first:

text

INSERT INTO leaderboard (user_id, score) VALUES ('mary1934', 1);

On subsequent calls, we'd just update their score:

text

UPDATE leaderboard set score=score + 1 where user_id='mary1934';

How do we find the top players of a leaderboard?

We can run the following query:

text

SELECT (@rownum := @rownum + 1) AS rank, user_id, score
FROM leaderboard
ORDER BY score DESC;

This is not performant though as it makes a table scan to order all records in the database table.

We can optimize it by adding an index on score and using the LIMIT operation to avoid scanning everything:

text

SELECT (@rownum := @rownum + 1) AS rank, user_id, score
FROM leaderboard
ORDER BY score DESC
LIMIT 10;

This approach, however, doesn't scale well if the user is not at the top of the leaderboard and you'd want to locate their rank.

#### Redis solution

We want to find a solution, which works well even for millions of players without having to fallback on complex database queries.

Redis is an in-memory data store, which is fast as it works in-memory and has a suitable data structure to serve our needs - sorted set.

A sorted set is a data structure similar to sets in programming languages, which allows you to keep a data structure sorted by a given criteria.
Internally, it is implemented using a hash-map to maintain mapping between key (user_id) and value (score) and a skip list which maps scores to users in sorted order:

How does a skip list work?
- It is a linked list which allows for fast search
- It consists of a sorted linked list and multi-level indexes

This structure enables us to quickly search for specific values when the data set is large enough.
In the example below (64 nodes), it requires traversing 62 nodes in a base linked list to find the given value and 11 nodes in the skip-list case:

Sorted sets are more performant than relational databases as the data is kept sorted at all times at the price of O(logN) add and find operation.

In contract, here's an example nested query we need to run to find the rank of a given user in a relational DB:

text

SELECT ,(SELECT COUNT() FROM leaderboard lb2
WHERE lb2.score >= lb1.score) RANK
FROM leaderboard lb1
WHERE lb1.user_id = {:user_id};

What operations do we need to operate our leaderboard in Redis?
- ZADD - insert the user into the set if they don't exist. Otherwise, update the score. O(logN) time complexity.
- ZINCRBY - increment the score of a user by given amount. If user doesn't exist, score starts at zero. O(logN) time complexity.
- ZRANGE/ZREVRANGE - fetch a range of users, sorted by their score. We can specify order (ASC/DESC), offset and result size. O(logN+M) time complexity where M is result size.
- ZRANK/ZREVRANK - Fetch the position (rank) of given user in ASC/DESC order. O(logN) time complexity.

What happens when a user scores a point?

text

ZINCRBY leaderboard_feb_2021 1 'mary1934'

There's a new leaderboard created every month while old ones are moved to historical storage.

What happens when a user fetches top 10 players?

text

ZREVRANGE leaderboard_feb_2021 0 9 WITHSCORES

Example result:

text

[(user2,score2),(user1,score1),(user5,score5)...]

What about user fetching their leaderboard position?

This can be easily achieved by the following query, given that we know a user's leaderboard position:

text

ZREVRANGE leaderboard_feb_2021 357 365

A user's position can be fetched using ZREVRANK <user-id>.

Let's explore what our storage requirements are:
- Assuming worst-case scenario of all 25mil MAU participating in the game for a given month
- ID is 24-character string and score is 16-bit integer, we need 26 bytes * 25mil = ~650MB of storage
- Even if we double the storage cost due to the overhead of the skip list, this would still easily fit in a modern redis cluster

Another non-functional requirement to consider is supporting 2500 updates per second. This is well within a single Redis server's capabilities.

Additional caveats:
- We can spin up a Redis replica to avoid losing data when a redis server crashes
- We can still leverage Redis persistence to not lose data in the event of a crash
- We'll need two supporting tables in MySQL to fetch user details such as username, display name, etc as well as store when eg a user won a game
- The second table in MySQL can be used to reconstruct leaderboard when there is an infrastructure failure
- As a small performance optimization, we could cache the user details of top 10 players as they'd be frequently accessed

---

Step 3: Design Deep Dive

To use a cloud provider or not

We can either choose to deploy and manage our own services or use a cloud provider to manage them for us.

If we choose to manage the services our selves, we'll use redis for leaderboard data, mysql for user profile and potentially a cache for user profile if we want to scale the database:

Alternatively, we could use cloud offerings to manage a lot of the services for us. For example, we can use AWS API Gateway to route API calls to AWS Lambda functions:

AWS Lambda enables us to run code without managing or provisioning servers ourselves. It runs only when needed and scales automatically.

Exmaple us

---

Readme (Readme.md) Copy Section

System Design Interview - An Insider's Guide (Vol 1 and 2)

Additonal Resources

Rate Limiting Copy Section

Consistent Hashing Copy Section

Key-Value Store Copy Section

Unique-ID Generator Copy Section

Web Crawler Copy Section

Chat Systems Copy Section

Search Autocomplete Copy Section

Youtube Copy Section

Google Drive Copy Section

01. Scaling/Readme (01. Scaling/Readme.md) Copy Section

Chapter 1: Scale from Zero to Millions of Users

Introduction Copy Chapter

Section 1: Single Server Setup Copy Chapter

Request Flow Copy Section

Traffic Sources Copy Section

Section 2: Database Separation Copy Chapter

Database Choices Copy Section

Section 3: Vertical vs Horizontal Scaling Copy Chapter

Vertical Scaling Copy Section

Horizontal Scaling Copy Section

Section 4: Load Balancer Copy Chapter

Section 5: Database Replication Copy Chapter

Master-Slave Model Copy Section

Benefits Copy Section

Failure Handling Copy Section

Section 6: Caching Copy Chapter

Caching considerations Copy Section

Section 7: Content Delivery Network (CDN) Copy Chapter

Workflow Copy Section

CDN considerations Copy Section

Section 8: Stateless Web Tier Copy Chapter

Section 9: Multi-Data Center Setup Copy Chapter

Key considerations Copy Section

Section 10: Message Queue Copy Chapter

Section 11: Logging, Metrics, and Automation Copy Chapter

Importance Copy Section

Section 12: Database Scaling Copy Chapter

Vertical Scaling Copy Section

Horizontal Scaling (Sharding) Copy Section

Conclusion Copy Chapter

Key Takeaways Copy Section

02. Back Of The Envelope Estimation/Readme (02. Back Of the Envelope Estimation/Readme.md) Copy Section

Chapter 2: Back-of-the-Envelope Estimation

Introduction Copy Chapter

Section 1: Key Concepts Copy Chapter

Power of Two Copy Section

Latency Numbers Every Programmer Should Know Copy Section

Availability Numbers Copy Section

Section 2: Example Estimation - Twitter QPS and Storage Requirements Copy Chapter

Assumptions Copy Section

Estimations Copy Section

Section 3: Tips for Effective Estimation Copy Chapter

1. Rounding and Approximation Copy Section

2. Write Down Assumptions Copy Section

3. Label Units Copy Section

4. Common Estimation Scenarios Copy Section

03. System Design Framework/Readme (03. System Design Framework/Readme.md) Copy Section

Chapter 3: A Framework for System Design Interviews

Introduction Copy Chapter

Step 1: Understand the Problem and Establish Design Scope Copy Chapter

Key Objectives Copy Section

Approach Copy Section

Example Copy Section

Step 2: Propose High-Level Design and Get Buy-In Copy Chapter

Key Objectives Copy Section

Approach Copy Section

Example Copy Section

Step 3: Design Deep Dive Copy Chapter

Key Objectives Copy Section

Approach Copy Section

Example Topics Copy Section

Step 4: Wrap-Up Copy Chapter

Key Objectives Copy Section

Approach Copy Section

Best Practices Copy Chapter

Dos Copy Section

Don’ts Copy Section

Readme (Readme.md)

Rate Limiting

Consistent Hashing

Key-Value Store

Unique-ID Generator

Web Crawler

Chat Systems

Search Autocomplete

Youtube

Google Drive

01. Scaling/Readme (01. Scaling/Readme.md)

Introduction

Section 1: Single Server Setup

Request Flow

Traffic Sources

Section 2: Database Separation

Database Choices

Section 3: Vertical vs Horizontal Scaling

Vertical Scaling

Horizontal Scaling

Section 4: Load Balancer

Section 5: Database Replication

Master-Slave Model

Benefits

Failure Handling

Section 6: Caching

Caching considerations

Section 7: Content Delivery Network (CDN)

Workflow

CDN considerations

Section 8: Stateless Web Tier

Section 9: Multi-Data Center Setup

Key considerations

Section 10: Message Queue

Section 11: Logging, Metrics, and Automation

Importance

Section 12: Database Scaling

Vertical Scaling

Horizontal Scaling (Sharding)

Conclusion

Key Takeaways

02. Back Of The Envelope Estimation/Readme (02. Back Of the Envelope Estimation/Readme.md)

Introduction

Section 1: Key Concepts

Power of Two

Latency Numbers Every Programmer Should Know

Availability Numbers

Section 2: Example Estimation - Twitter QPS and Storage Requirements

Assumptions

Estimations

Section 3: Tips for Effective Estimation

1. Rounding and Approximation

2. Write Down Assumptions

3. Label Units

4. Common Estimation Scenarios

03. System Design Framework/Readme (03. System Design Framework/Readme.md)

Introduction

Step 1: Understand the Problem and Establish Design Scope

Key Objectives

Approach

Example

Step 2: Propose High-Level Design and Get Buy-In

Key Objectives

Approach

Example

Step 3: Design Deep Dive

Key Objectives

Approach

Example Topics

Step 4: Wrap-Up

Key Objectives

Approach

Best Practices

Dos

Don’ts

Time Management

Suggested Time Allocation (for 45-Minute Interviews):

04. Rate Limiter/Readme (04. Rate Limiter/Readme.md)

Introduction

Benefits of Rate Limiting