Distributed Email Service

Design a large-scale email service (Gmail, Outlook, Yahoo Mail). In 2020, Gmail had 1.8B+ active users, Outlook 400M+.

Figure 1: Popular email providers

Step 1 of 4: Understand the problem and establish design scope #

Goal: lock in requirements and scale.

Requirements #

1. 1 billion users
2. Features: send/receive emails, fetch all emails, filter by read/unread, search by subject/sender/body, anti-spam/anti-virus
3. Authentication out of scope
4. HTTP for client-server communication (not SMTP/POP/IMAP for clients)
5. Attachments supported

Non-functional:

1. Reliability: no email data loss
2. Availability: auto-replication across nodes, partial failures tolerated
3. Scalability: handle growing users/emails without performance degradation
4. Flexibility/extensibility: custom protocols may be needed (POP/IMAP limited)

Back-of-the-envelope estimation:

1. 1B users, avg 10 sent/day → QPS = 10⁹ × 10 / 86,400 = 100,000
2. Avg 40 received/day, metadata 50KB → yearly metadata: 1B × 40 × 365 × 50KB = 730 PB
3. 20% of emails have attachments, avg 500KB → yearly attachments: 1B × 40 × 365 × 20% × 500KB = 1,460 PB
4. → Distributed database required

Step 2 of 4: Propose high-level design and get buy-in #

Goal: understand protocols, then the distributed architecture + send/receive flows.

Email knowledge 101 #

Protocols:

1. SMTP: standard for sending between mail servers
2. POP: download from server to local, deletes from server (single device access). Downloads the entire email (slow for large attachments). RFC 1939
3. IMAP: emails stay on the server, multi-device access. Downloads only headers until opened. Most widely used
4. HTTPS: used for web-based email (e.g. Outlook ActiveSync for mobile)

DNS: MX records for the recipient domain. Lower priority number = more preferred.

Figure 2: MX records

Attachments: Base64 encoding, MIME specification. Size limits: Outlook 20MB, Gmail 25MB.

Traditional mail servers #

Figure 3: Traditional mail servers

Flow: Alice (Outlook) → SMTP → Outlook server → DNS lookup → SMTP → Gmail server → Bob fetches via IMAP/POP.

Storage: historically Maildir: each email a separate file in the user directory. Works for small scale; disk I/O becomes a bottleneck for billions of emails. No HA/reliability.

Figure 4: Maildir

Distributed mail servers #

Email APIs (RESTful/HTTP for webmail):

1. POST /v1/messages: send email (To, Cc, Bcc)
2. GET /v1/folders: returns all folders. RFC6154 defaults: All, Archive, Drafts, Flagged, Junk, Sent, Trash
3. GET /v1/folders/{folder_id}/messages: messages under a folder (supports consecutive + range-based paging)
4. GET /v1/messages/{message_id}: message details (from, to, subject, body, is_read)

Distributed mail server architecture:

Figure 5: High-level design

Components:

1. Web servers: public-facing, handle all API requests (login, signup, email CRUD)
2. Real-time servers: push new email updates via WebSocket (fallback: long polling). Stateful (persistent connections). Apache James implements JMAP over WebSocket
3. Metadata database: stores subject, body, from/to. Discussed in the deep dive
4. Attachment store: object storage (S3). Not Cassandra (blob limit <1MB practical, row cache issues with large attachments)
5. Distributed cache (Redis): recent emails cached in memory. Redis offers lists, easy to scale
6. Search store: distributed document store with inverted index for full-text search

Email sending flow:

Figure 6: Email sending flow

1. User writes email → load balancer (rate limiting)
2. Web server validates (size limits, etc.)
3. If same domain (sender = recipient): insert directly to storage/cache/object store, skip SMTP
4. Message queue: valid → outgoing queue; invalid → error queue
5. SMTP outgoing workers: pull from queue, check spam/virus, store in "Sent Folder", send to recipient server

The message queue decouples web servers from SMTP workers → independent scaling. Monitor queue size:

1. Recipient server unavailable → exponential backoff retry
2. Not enough consumers → add more workers

Email receiving flow:

Figure 7: Email receiving flow

1. Incoming → SMTP load balancer
2. SMTP servers: apply acceptance policy, bounce invalid emails
3. Large attachments → stored in S3 (not queued)
4. Emails → incoming email queue (buffer for surges)
5. Mail processing workers: spam filtering, virus scanning
6. Valid email → stored in mail storage, cache, object store
7. Receiver online → push to real-time servers (WebSocket)
8. Receiver offline → stored in the storage layer; on reconnect, webmail client fetches via RESTful API

Step 3 of 4: Design deep dive #

Goal: the metadata DB, deliverability, search.

Metadata database #

Characteristics of email metadata:

1. Headers: small, frequently accessed
2. Body: variable size, infrequently accessed (read once)
3. Operations isolated to an individual user
4. 82% of read queries are for data <16 days old
5. High reliability (no data loss)

Database options:

Custom database characteristics needed: single column up to MB size, strong consistency, reduced disk I/O, HA/fault-tolerant, easy incremental backups.

Data model: user_id as partition key (one user's data on one shard). Messages not shared across users (not required).

Queries & tables:

Query 1. Get all folders: partition key user_id. All folders for a user in one partition.

Table 1: Folders by user

Query 2. Display emails for a folder: composite PK <user_id, folder_id>. Clustering key: email_id (TIMEUUID, chronological order).

Table 2: Emails by folder

Query 3. Get email details: SELECT * FROM emails_by_user WHERE email_id = 123;. Attachments retrieved by <email_id, filename>.

Table 3: Emails by user

Query 4. Fetch read/unread emails: problem, is_read is neither a partition nor a clustering key → NoSQL rejects the query. Solution: denormalization into two tables: read_emails and unread_emails. Marking UNREAD→READ: delete from unread, insert into read.

Table 4: Read and unread emails

Bonus: Conversation threads: uses headers Message-Id, In-Reply-To, References. The JWZ algorithm reconstructs conversations from reply chains.

Consistency trade-off: single primary per mailbox. On failover, the mailbox is inaccessible until failover completes. Trades availability for consistency (correctness is paramount).

Email deliverability #

50% of all emails are spam. New mail servers have no reputation: their emails likely land in spam.

Best practices:

1. Dedicated IPs: separate IPs for sending; history builds reputation
2. Classify emails: different IPs for marketing vs. important mail
3. Warm-up: 2-6 weeks to build reputation with major providers (Office365, Gmail, Yahoo)
4. Ban spammers quickly before reputation damage
5. Retries: max threshold → store in queue → monitor → investigate
6. Feedback processing:

Figure 8: Handle feedback loop

Separate queues for each type.

7. Email authentication: SPF, DKIM, DMARC combat phishing (93% of breaches are phishing/pretexting)

Figure 9: Gmail header example (SPF, DKIM, DMARC)

Search #

Email search vs Google search:

Write-heavy (reindex on send/receive/delete). Read occurs only on explicit search.

Option 1: Elasticsearch

Figure 10: Elasticsearch

1. Partition by user_id to group a user's docs on the same node
2. Search: synchronous. Reindexing: offline via Kafka
3. Challenge: keeping the primary store in sync with Elasticsearch

Option 2: Custom search engine

1. Main bottleneck: disk I/O (PB-level daily metadata)
2. Use an LSM tree (Log-Structured Merge-Tree) for the index: write path optimized via sequential writes. New email → level 0 in-memory cache → threshold → merge to the next level
3. Separate frequently changing data (folders) from static data (emails)

Figure 11: LSM tree

Comparison:

Guideline: small scale → Elasticsearch; Gmail/Outlook scale → custom native search embedded in the DB.

Scalability and availability #

Data replicated across multiple data centers. Users connect to the nearest server. Network partition → access from other DCs.

Figure 12: Multi-data center setup

Step 4 of 4: Wrap up #

What we designed:

1. Requirements → estimation → traditional vs distributed servers
2. APIs → sending/receiving flows
3. Metadata DB (NoSQL with denormalization)
4. Deliverability
5. Search (Elasticsearch vs LSM-based custom)
6. Multi-DC scalability

Additional topics:

1. Fault tolerance (node/network failures)
2. Compliance (GDPR, PII, legal intercept)
3. Security (phishing protection, encryption)
4. Optimization (deduplicate attachments in group emails)