Designing Facebook Messenger

1. What we are building

Facebook Messenger is a real-time, text-based instant-messaging service. Users chat one-on-one (and in groups) from phones and the web, see who is online, and expect their full chat history to be present and identical on every device they log in from.

The design splits into three hard problems that pull against each other: real-time delivery (get a message to the recipient with minimum latency), durable, consistent history (never lose a message; show the same transcript everywhere), and presence (track online/offline for hundreds of millions of connections without melting the fleet).

Requirements

Functional

• Support one-on-one conversations between users.
• Track online/offline status of users.
• Persist chat history durably.

Non-functional

• Real-time experience with minimum latency.
• Strong consistency of history: the same transcript on all of a user's devices.
• High availability is desirable, but we prefer consistency when the two conflict.

Extended

• Group chats.
• Push notifications to reach users while they are offline.

2. Capacity estimation

Rough numbers are what drive the storage-engine and sharding choices later, so we do them first.

Every inbound message must go back out to a recipient, so we need roughly the same 25 MB/s for both ingress and egress. The 3.65 PB figure excludes user records, per-message metadata (ID, timestamp, sequence number), compression, and replication — real footprint is a few multiples of this. The takeaway that matters for design: this is a write-heavy stream of tiny records that we mostly read back sequentially per conversation.

3. Delivery: push, not pull

To receive a message the client can pull (poll the server periodically) or push (keep a connection open and let the server deliver instantly).

Pull forces a latency-vs-waste tradeoff: poll often and you burn resources on mostly-empty responses; poll rarely and messages feel laggy. The server also has to keep buffering undelivered messages until the next poll.

Push wins for chat. Active users hold a connection open; the instant the server has a message it writes it down the existing connection — minimum latency, no polling storm, and no need to track pending messages for online users.

Holding the connection: long-poll vs WebSocket

Two ways to keep a channel open:

• HTTP long polling — the client sends a request; the server holds it open until it has data (instead of returning an empty response), then replies and the client immediately re-requests. Simple, proxy/firewall-friendly, but each message costs a full request/response round and connections silently time out, forcing reconnects.
• WebSockets — one upgraded TCP connection carries full-duplex frames for the session's lifetime. Lower per-message overhead and true bidirectional push; the cost is stateful long-lived sockets, heavier load-balancer/proxy handling, and explicit heartbeat/reconnect logic.

Modern Messenger-class systems use WebSockets (with a long-poll fallback for hostile networks). Either way the server must map each connected user to the box holding their socket.

4. Connection registry, ordering, and offline handling

Routing to an open socket. Each chat server keeps a hash table UserID → connection. On receiving a message for a user it looks up the connection and writes on it. Fleet sizing: plan for 500 M concurrent connections; if one modern box handles ~50K, that's ~10K chat servers. A load balancer maps each UserID to the server that currently owns its connection.

Processing a send. On a new message the server (1) appends it durably to the message store — an LSM memstore write plus its WAL append, a cheap sequential I/O — then (2) acks the sender as sent, and (3) delivers it to the recipient's server, which later produces the separate delivered receipt. The tempting shortcut — ack first, store asynchronously in the background — is a durability bug under this page's own requirements: we declared "never lose a message" and consistency-over-availability, yet a chat-server crash after an early ack silently drops an acknowledged message. So the durable (WAL) write sits on the ack path, costing a few milliseconds of sender latency; everything after durability (replica catch-up, cache fill, receipts) stays asynchronous.

Ordering. A server-receive timestamp is not enough: if A→B arrives at T1 and B→A arrives at T2>T1, ordering by timestamp makes each participant see a different-but-plausible sequence, and worse, isn't stable across a user's own devices. The fix is a per-user monotonic sequence number: every message is stamped with the next sequence number in that user's own stream. Two participants may still see slightly different interleavings of a conversation, but each user's view is identical on all of their devices — which is exactly the consistency requirement. (Section 6 shows why UserID-based sharding makes this sequence number cheap to generate.)

Offline recipient. If the recipient is genuinely gone, notify the sender of delivery failure. If it's a transient drop (a long-poll timeout, a flaky socket), expect a reconnect: buffer the message briefly and retry on reconnect, and/or let the client re-send transparently so the user never retypes.

5. Storage engine: why a wide-column store

The access pattern is a firehose of tiny writes plus per-conversation range reads ("give me the last N messages, then page backward"). That immediately rules some options out and points to one family.

• Not a plain RDBMS (MySQL/Postgres) as the message store. A row read/write per message at 20 B/day, plus B-tree index maintenance on a write-dominated table, is high-latency and crushing load. RDBMS remains the right home for user accounts and relationships, where you need joins and transactions.
• Not a document store (MongoDB) for the hot message path either — same per-document write/read cost and index pressure for our shape.
• A wide-column store fits. HBase — modeled on Google's Bigtable, running over HDFS — buffers writes in an in-memory memstore and flushes sorted files to disk (an LSM design). That absorbs huge volumes of small writes cheaply and, because data is stored sorted by key, serves range scans ("messages in conversation X, newest first") efficiently. It also stores variable-sized values well. Cassandra and ScyllaDB are the same wide-column, LSM-based family and are equally valid choices.

Real-world corroboration. Discord's message store evolved through exactly this reasoning. Per Discord's own engineering blogs, it began on MongoDB, migrated to Cassandra in 2017 as message volume exploded, and then re-platformed onto ScyllaDB in 2023 (a C++ rewrite of the Cassandra data model) to tame tail latency and GC pauses at trillions of messages. The throughline matches our choice: a document store did not hold up for a write-heavy chat firehose, and the endpoint is a wide-column, LSM store — the same category as HBase.

6. Data partitioning

At 3.65 PB over five years we must shard. Two candidate keys:

Shard by UserID (chosen). Hash the UserID and keep all of a user's messages — their entire inbox across every conversation — on one shard: shard = hash(UserID) % 1000. At ~4 TB/shard, 3.65 PB needs ~900 shards; round to 1K logical shards, initially packed several-per-physical-server and spread out as data grows. Fetching a user's history is then a single-shard operation.

Shard by MessageID (rejected). Scattering one conversation's messages across shards turns every history fetch into a slow cross-shard scatter-gather. Don't.

Reconciling sharding with the per-user sequence number

These two decisions are not in tension — one enables the other. Because UserID sharding co-locates a user's entire message stream on a single home shard, the per-user monotonic sequence number from Section 4 can be produced by a single counter on that one shard. No cross-shard coordination, no distributed consensus, no clock sync is needed to hand out the next number — the very property (all my data on one shard) that makes history reads fast is what makes strictly-ordered, device-consistent sequencing cheap. Sharding by MessageID would have scattered a user's stream and made that counter a distributed-coordination problem.

7. Presence, cache, balancing, fault tolerance, and extensions

Presence (online/offline)

The connection object already tells us who is online. Broadcasting every flip to all friends of 500 M users would be ruinous, so we soften it: pull a friend list's statuses on app start; mark a user offline lazily when a send to them fails; delay broadcasting "came online" a few seconds to absorb flapping; and only pull status for users actually in the viewport, tolerating slightly stale offline state.

Cache

Cache the last ~15 messages of the last ~5 conversations in a user's viewport. Since a user's whole history lives on one shard, their cache can live entirely on one machine too.

Load balancing

A load balancer in front of the chat servers maps each UserID to the server holding its connection; a second balancer fronts the cache tier.

Fault tolerance & replication

Failing over live TCP connections is impractical — instead, clients auto-reconnect with jittered exponential backoff and the LB routes them to a new chat server. The jitter matters: a dead box means ~50K sockets re-authenticating and catch-up-syncing near-simultaneously (an AZ loss multiplies that by hundreds of servers), so spread the reconnects over tens of seconds and rate-limit the catch-up pull path to protect the message store during mass reconnects. Message data is never single-copy: replicate the hot message log across servers; erasure coding (e.g., Reed–Solomon) only pays off at the aggregated file/block layer (HDFS-style EC on flushed HFiles) — never per ~100-byte message, where the coding metadata would dwarf the payload.

Group chat

Model a group as a GroupChatID object holding its member list; the LB routes group messages by GroupChatID, and the owning server fans the message out to each member's connection server. Group chats are stored in their own table partitioned by GroupChatID.

Push notifications

Offline users are reached via a Notification server that forwards messages to the platform push services (APNs, FCM), which deliver to the device. This is what turns "send fails when recipient is offline" into "recipient gets a banner and syncs on next open."

8. The judgment layer: when to use each choice, and when NOT

Delivery model — push vs pull

Open channel — WebSocket vs long-poll

Message store — wide-column vs RDBMS vs document

Partition key — UserID vs MessageID

Bottom line: Messenger is a push system over WebSockets, backed by a wide-column message log sharded by UserID, with a per-user sequence number for device-consistent ordering, and push notifications to bridge the offline gap.

Sources

• Grokking the System Design Interview — "Designing Facebook Messenger" (problem statement, capacity math, HBase choice, UserID partitioning). Original lesson: site/system-design/system-design-problems/005-designing-facebook-messenger.html.
• Discord Engineering, "How Discord Stores Billions of Messages" (2017) — migration from MongoDB to Cassandra.
• Discord Engineering, "How Discord Stores Trillions of Messages" (2023) — migration from Cassandra to ScyllaDB.
• Apache HBase and Google Bigtable references (LSM / wide-column storage model).

🪜 Drill ladder: Designing Facebook Messenger

1. L1 — Group chat fan-out: a 200-member group, every member on a different chat server. One message arrives at the group's owning server — count the cross-server writes it triggers, and compare against fanning out at the sender's server instead. (Answer: ~200 either way, but only the GroupChatID-owner design keeps ordering per group on one box.)
2. L2 — Presence accuracy: 500M users averaging 300 friends, each flipping online/offline twice a day. Compute the eager-broadcast rate (500M × 2 × 300 ≈ 300B notifications/day ≈ 3.5M/s) and state which two of this page's softenings kill most of that load.
3. L3 — Reconnect stampede: one chat server dies holding 50K sockets. Compute the reconnect QPS at the LB if clients retry uniformly within 1s versus jittered over 30s (50K/s vs ~1.7K/s), and name what else must be rate-limited besides the handshake (catch-up sync reads).
4. L4 — Delivery receipts vs durability: with the WAL append on the ack path, estimate the added sender-perceived latency (~1–5ms sequential append) and state exactly what an ack-before-store design silently loses when the chat server crashes (acknowledged-but-unstored messages).