Loading...
[Mid-level engineer topic.]
Let’s perform a quick scale estimation to understand the challenges of building a storage-heavy email service.
1 billion users
Email sends: Assume each user sends 10 emails/day, resulting in an approximate QPS for sending emails of:
If you’re seeing write throughput exceeding the capabilities of a single server (around 10,000-100,000 writes per second for high-end database servers), a distributed system is necessary.
- Email receives: Each user receives 30 emails/day. Assuming email metadata (excluding attachments) averages 40 KB, the yearly storage for metadata would be:
- Attachments: Assume 25% of emails contain attachments, with an average size of 600 KB. The yearly storage for attachments would be:
A single database instance can typically handle between 500 GB to 2 TB of data efficiently, depending on the database engine and hardware.
If the data volume is expected to exceed 1-2 TB, it's a good time to consider sharding or moving to a distributed database system. In our case since our data is expected to be in the 1.6k PB range we will definitely consider using a distributed database system.
[Calculating storage here will help us decide and justify database choices]
POST /v1/messages: Handles email sending, targeting recipients in To, Cc, Bcc.
GET /v1/folders: Returns all folders associated with an email account (Inbox, Drafts, Trash, etc.).
GET /v1/folders/{folder_id}/messages: Fetches all emails in a folder (supports pagination for large datasets).
GET /v1/messages/{message_id}: Returns metadata and content of a specific email.
GET /v1/search: Perform full-text search across email bodies, headers, and metadata.
[This should be kept as simple as possible. Just list the essential features]
Relational Database (e.g. MySQL) is a solid relational database, but it relies on vertical scaling, meaning it scales by adding more hardware, which becomes expensive and inefficient as data grows.
Handling large, unstructured data like email bodies and attachments, along with complex queries such as full-text search, is a challenge for Relational Databases.
At 100,000 QPS for writes, Relational Database would likely hit performance bottlenecks and degrade under heavy loads.
MongoDB, being a document-based NoSQL system, is more flexible for semi-structured data, making it easier to store emails and attachments.
MongoDB performs well for read-heavy workloads and supports more advanced queries like range and full-text search out of the box.
However, for write-heavy workloads, MongoDB's journaling and locking mechanisms can introduce overhead, especially when writes are spread across multiple shards.
While it supports horizontal scaling, it doesn’t scale as efficiently as Cassandra for continuous, high-volume writes.
This brings us to Cassandra, which is optimized for write-heavy systems.
Its log-structured merge (LSM) tree architecture handles high write throughput without bottlenecks, making it ideal for storing millions of emails per second.
Although Cassandra isn’t as strong for complex queries or full-text search, pairing it with a system like Elasticsearch can handle that need.
Cassandra's peer-to-peer architecture allows for near-infinite horizontal scaling with no single point of failure, making it perfect for large-scale email systems.
In short, MySQL can't scale efficiently, MongoDB handles reads well but struggles with heavy writes, and Cassandra is the best fit for handling 100,000+ QPS in a write-intensive, distributed system. Its scalability and performance make it the most suitable option for our needs.
[At the scale of Gmail or Outlook, database systems are typically custom-made to optimize for high IOPS, since input/output constraints become a significant challenge. During an interview it's probably not realistic to use a custom-build solution. However, we can still raise this point and suggest using an existing database that we are likely more familiar with.]
Partition Key: This is critical in Cassandra because it determines how data is distributed across the nodes. In the email system context:
Partition key: user_id
This ensures that all emails, folders, and metadata related to a specific user are stored in the same partition, allowing efficient queries on a per-user basis.
Clustering Key: Clustering keys define how data within a partition is ordered. For example:
Clustering key: folder_id and email_id (TIMEUUID)
These keys ensure that emails within a user’s folder are sorted by their timestamp (email_id as TIMEUUID), making it easy to retrieve emails in chronological order.
This table would store emails organized by user and folder:
CREATE TABLE emails_by_folder (
user_id UUID, -- Partition Key
folder_id UUID, -- Clustering Key
email_id TIMEUUID, -- Clustering Key (sorted by timestamp)
from TEXT,
subject TEXT,
preview TEXT,
is_read BOOLEAN,
PRIMARY KEY (user_id, folder_id, email_id)
);
Since emails can have multiple attachments, we might want to separate attachments into their own table:
CREATE TABLE email_attachments (
email_id TIMEUUID, -- Partition Key
filename TEXT, -- Clustering Key
size INT,
url TEXT,
PRIMARY KEY (email_id, filename)
);
Denormalization is key in Cassandra because it helps optimize query performance. Since Cassandra can’t filter on non-partitioned keys or perform complex joins, you need to duplicate data across multiple tables to handle different queries efficiently.
To efficiently query read and unread emails, the system needs separate tables for read and unread emails. This denormalization allows for fast filtering based on the read status of an email:
CREATE TABLE unread_emails (
user_id UUID, -- Partition Key
folder_id UUID, -- Clustering Key
email_id TIMEUUID, -- Clustering Key (sorted by timestamp)
from TEXT,
subject TEXT,
preview TEXT,
PRIMARY KEY (user_id, folder_id, email_id)
);
CREATE TABLE read_emails (
user_id UUID, -- Partition Key
folder_id UUID, -- Clustering Key
email_id TIMEUUID, -- Clustering Key (sorted by timestamp)
from TEXT,
subject TEXT,
preview TEXT,
PRIMARY KEY (user_id, folder_id, email_id)
);
When an email is marked as "read," it is deleted from the unread_emails table and inserted into the read_emails table. This duplication makes it efficient to fetch all unread or read emails without needing to filter on the is_read flag within a single table, which would not be performant in Cassandra.
Conversation threads can be handled using email headers (like Message-ID, In-Reply-To, and References). Since Cassandra doesn’t support joins, we might need to precompute and store conversations separately if efficient thread retrieval is required.
CREATE TABLE conversation_threads (
user_id UUID, -- Partition Key
conversation_id UUID, -- Clustering Key
email_id TIMEUUID, -- Clustering Key (sorted by timestamp)
subject TEXT,
PRIMARY KEY (user_id, conversation_id, email_id)
);
Emails related to a conversation are grouped by conversation_id. This allows efficient retrieval of all emails in a thread, ordered by email_id (timestamp).
[During an interview you do not need to list out every table like this. We recommend that you list out just one and touch points on these concepts to save time.]
In Cassandra, strong consistency across multiple nodes is not always guaranteed unless configured using quorum reads/writes. For an email system, eventual consistency may be acceptable for certain parts of the system (e.g., email attachments), while strong consistency can be enforced for critical data like email metadata by tuning the replication and consistency levels.
Cassandra’s multi-master architecture ensures high availability, but this can lead to consistency trade-offs. The system would likely be configured to prioritize consistency for email metadata while using eventual consistency for less critical parts, such as email bodies or attachments.
Users interact with the system via web browsers, mobile apps, or desktop clients. These clients use web APIs to send, receive, and search emails.
For traditional email clients, standard protocols like SMTP (for sending), IMAP, and POP (for receiving) are used.
Modern clients, especially web and mobile apps, use RESTful APIs for operations like composing, sending, and reading emails.
These handle all requests from the webmail clients, including user login, email retrieval, and email sending.
Basic email validation, user session management, ensuring email is virus-free, managing rate limits, and routing requests to the right backend services.
All API requests for email actions (such as sending emails, fetching folders, loading emails in a folder) pass through these web servers.
These servers manage real-time email notifications to clients using WebSockets or long-polling as a fallback. This ensures that new emails are pushed to users in real-time when they are online.
Real-time servers maintain persistent connections with the client for live updates.
Metadata Database stores the email metadata, including sender/recipient info, subject, and email body (excluding attachments). The metadata database is built to efficiently handle large amounts of data while offering fast queries (e.g., for sorting emails by time or searching email subjects).
Large email attachments (e.g., images, files) are stored separately from the main metadata database, usually in distributed object stores like Amazon S3. This ensures attachments can scale independently.
A cache (e.g., Redis) is used to store the most recent emails or frequently accessed emails, reducing load on the main database and improving response times for users.
Full-text search functionality is powered by an inverted index stored in a dedicated search store (e.g., Elasticsearch). This allows fast retrieval of emails based on keywords or other criteria.
The design incorporates message queues (e.g., Kafka) to decouple email sending and receiving from the immediate user interaction. This ensures email processing (e.g., virus checking, spam filtering) is handled asynchronously and scales independently.
Outgoing Queue: Emails that pass validation are queued for delivery.
Incoming Queue: Incoming emails are queued for processing before being stored and delivered to users.
When a user sends an email:
When an email is received:
Email search is different from web search because it's limited to a user’s mailbox and requires near-instant, accurate results.
For very large systems, custom search engines are sometimes used, optimized with structures like Log-Structured Merge Trees (LSM) to improve indexing efficiency. This helps reduce disk I/O and handle the scale of reindexing effectively.
However, we will focus on using Elasticsearch as a solution.
Each time an email is sent, received, or deleted, the system must reindex the data, leading to more writes than reads.
Elasticsearch is a common solution for full-text search, but syncing it with the main database adds complexity.
To integrate Elasticsearch, emails are indexed when created or received, extracting metadata like sender, recipient, subject, and body.
Reindexing happens whenever an email is updated, marked as read, or flagged, and Kafka can be used to handle these frequent writes asynchronously, preventing performance bottlenecks.
The index is structured by user_id, allowing efficient searches within a user’s mailbox, with fields like email_id, subject, body, and status.
As the system scales, Elasticsearch can be horizontally scaled by adding more nodes, and synchronization with the primary datastore is crucial to ensure consistency.
Use dedicated IP addresses for different categories of emails (e.g., transactional vs. promotional) helps build and maintain a strong sending reputation, reducing the likelihood of emails being flagged as spam.
New IP addresses must be "warmed up" gradually to avoid being blacklisted by major providers like Gmail or Outlook.
Spammers should be quickly identified and blocked to prevent damage to the server's reputation.
The system is designed to scale horizontally, meaning that adding more servers can accommodate increasing demand.
Users' data access patterns are largely independent, making it easy to scale most components horizontally.
Data is replicated across multiple data centers to ensure high availability and fault tolerance.
See Database section on Trade offs between different types of databases.
In the event of a network partition, users can still access their emails from other data centers.
The system also employs message queues to decouple sending and receiving email operations, allowing email traffic to be processed asynchronously and preventing web servers from being overwhelmed during peak loads.
We can look into introducing machine learning-based algorithms for smarter email classification.