Hotshard

Design a Chat System (WhatsApp / Messenger)

Let a billion people send messages to each other and to groups, and have each message show up on the other person's phone in a fraction of a second — even when they are offline right now.

Predict ~1.5 h · Read ~55 min

The problem i

A chat system carries messages between people. You type a message, and it appears on your friend's phone almost instantly. If they are not online right now, it is waiting for them the moment they open the app. That is the whole product from the outside: a message goes in on one phone and comes out on another, fast, and nothing gets lost. Group chats are the same idea with more than one recipient.

The thing that makes this hard is not storing the messages — it is reaching the recipient. A message is tiny, but at our scale there are about a billion people using the app, and each one who is online is connected to some server. When you send a message, the system has to get it from wherever you are to wherever your friend is. That routing problem is the center of this design.

The load is real: about a hundred billion messages a day, which is roughly a million a second on average and about three million a second at peak. Most messages go to one person or a small group, but the tail matters — a group can have up to a thousand members, and one message to a thousand-member group is a thousand deliveries. We design for both the common case and that tail.

To stay focused, we leave a few things out. We do not design the end-to-end encryption — how two phones agree on secret keys is a cryptography problem on its own, not a decision about which server or database to use, so it does not change the shape of this design. Everything we keep is about one thing: getting a small message from one phone to another, quickly and reliably, at this scale.

Functional requirements what it must do

  • Send a message to one recipient — the core write, small (about a kilobyte) but arriving at millions a second.
  • Deliver to a recipient who is online in well under a second — the target: for ninety-five of every hundred messages, under half a second.
  • Deliver to a recipient who is offline, whenever they come back.
  • Send a message to a group — a one-to-one send is one delivery, a large group is up to a thousand.
  • Show delivery and read status — sent, delivered, read.
  • Keep the conversation history.

Non-functional requirements what it must be

  • Sending feels instant — the sender is told the message is safely stored in about two hundred milliseconds.
  • Messages arrive in order within a chat — a reply never renders above the message it answers.
  • The service survives one of its servers dying.
  • No message is silently lost.

Out of scope left out on purpose

  • End-to-end encryption
  • Voice and video calls
  • Spam and abuse filtering
  • The social graph (contacts, friend requests)

How the design works

The finished design, explained one piece at a time: first the contract the system serves, then the smallest version that works, then the version that survives scale — walked use case by use case, one deep dive at a time, with the finished diagram at the end.

The API

The whole product is these endpoints. Every box in the design below exists to serve one of them.

WS /connectOpen one live connection and keep it open. Everything travels over this one connection: you send messages up it, and new messages are pushed down it the instant they exist, so the phone never has to ask. How this connection works is the first thing the design below settles.
contract ▾
GET /connect
  Upgrade: websocket
  Authorization: Bearer <token>
 101 Switching Protocols        # the connection is now open, held by one server

# frames after upgrade (JSON), both directions on the same connection:
 { "t": "ping" }                # heartbeat, every 10–30s
 { "t": "pong" }

# send a message up the connection; the reply says it is safely stored:
 { "t": "send", "chat_id": "c_92f", "client_msg_id": "uuid-for-dedup", "body": "hi" }
 { "t": "ack",  "chat_id": "c_92f", "client_msg_id": "uuid-for-dedup", "seq": 4181, "status": "stored" }

# messages and status for you arrive pushed down the same connection:
 { "t": "msg",     "chat_id": "...", "seq": 4182, "from": "...", "body": "..." }
 { "t": "receipt", "chat_id": "...", "seq": 4181, "state": "read" }
GET /messages?chat_id&after_seqCatch up. A phone that was offline or has just reconnected asks for everything it missed in a chat since the last message it saw — exact, with no gaps and no duplicates.
contract ▾
GET /messages?chat_id=c_92f&after_seq=4180&limit=200
 200 OK
  { "messages": [ { "seq": 4181, ... }, { "seq": 4182, ... } ],
    "next_after_seq": 4182 }     # the client stores this per chat and resumes from it

part 1 The single-server version

On day one, this is the whole system: one app server and one database. Your phone holds no special connection to it. It just makes ordinary web requests, the same way a browser loads a page.

A few words we will use throughout. A message is one small piece of text — about a kilobyte once you count who sent it, when, and which chat it belongs to. A chat is a conversation. It can be between two people or among a group, and every message belongs to exactly one chat. To deliver a message means to get it onto the recipient's screen. Notice that storing a message and delivering it are two different jobs. On day one they feel like the same thing. At scale they come apart, and most of this design lives in that gap.

The single server does two things. When you send, your phone posts the message to the server. The server writes one row into the database and replies 'stored.' Receiving is clumsier. Your friend's phone has no way to know a new message exists, so it asks. It asks every few seconds, over and over: 'anything new in my chats?' That pattern is called polling, and it is how this version delivers messages.

phoneserver×1database
The whole system on day one.
Following a request
Send a message
  1. the phone posts the message to the server; separately, it polls every few seconds to ask if anything new arrivedtens of ms
  2. the server writes one row into the database and replies 'stored' — one durable write~10 ms

One hop dominates every path: the client's request crossing the public internet to reach the datacenter. That hop is tens of milliseconds — call it about 30 — and it is set by physical distance, not by anything we build. Everything after it happens inside a single datacenter, where the network between two machines is well under a millisecond — so those numbers are the actual work, not the wait: a durable write to disk about 10 ms, an indexed read about 5 ms. A hop between two datacenters would cost 10–100× the in-datacenter number, which is why the whole path stays in one region.

Polling is what breaks first. Suppose every phone asks every few seconds. With hundreds of millions of phones online, the server is answering tens of millions of these requests every second — and almost every answer is 'no, nothing new.' You pay the full cost of a request and get an empty reply. Even at that price, a new message just sits on the server until the next poll comes around, so delivery never feels instant. The database has its own problem: a single node cannot absorb three million real writes a second.

So the single-server version is fine for a thousand users and falls over at a billion. The rest of this design builds the real system one capability at a time. Each use case adds the smallest next piece the system needs.

part 2 The version that survives scale

Before any new boxes, one reframe that makes the whole design click. Chat is not really a storage problem. The messages are tiny, and any serious store can hold them. Chat is a routing problem. The recipient is a moving target: they are connected to some server right now, and tomorrow they will be connected to a different one. The sender's side has to find them, fast, every single time.

We build the system in the order you would actually build it. Get a phone online. Route one message. Keep messages safe and in order. Deliver to someone who is offline. Handle a big group. Show receipts. Survive a server dying. Each use case forces exactly one new piece of the machine, so the diagram grows in front of you — you never meet a box before the use case that needs it.

Two ways through this section

The eleven decisions below are the heart of this design, and there are two ways to take them. In the design room you make each one yourself and the board reacts to your pick. The write-ups after this point cover the same decisions in full — read them instead, or afterwards as reference.

use case 1 Getting online

Nothing works until a phone can be reached. So the first use case is just that: your phone comes online, ready to receive, though no message is sent yet. Two questions decide it — what 'online' physically means for one phone, and what it costs to hold two hundred million of them in that state at once.

deep dive 1 The realtime channel
The question

'Online' here means one narrow thing: your own phone can be reached by a server, right now. (Seeing which of your friends are online is a different feature, and we leave it for the end.) So the question is: what kind of channel lets a server put a message on your phone's screen the moment that message exists, without the phone asking for it?

phoneserver
polling — the phone keeps asking
anything new?
no
anything new?
no
a message for you arrives at the server…
anything new?
yes — here it is (late)
long-polling — ask, then wait
anything new? (server holds the request open)
…a message arrives…
here it is — now reconnect and ask again
WebSocket — one connection, held open
open connection (upgrade) — stays open
…a message arrives…
pushed down the open socket, instantly
Three ways to get a message onto a phone

There are three honest ways to get a message onto a phone: keep asking, ask and wait, or hold a line open.

Keep asking is what the day-one version already does — polling, just turned up. It has a wall built into it. To make delivery feel instant, the phone would have to ask many times a second, and almost every answer would still be 'nothing new.' So the request rate explodes while the useful work stays the same. And even then, there is always a gap between polls. You can pay more and more and never actually reach 'instant.'

Ask and wait is smarter. The phone sends a request, and the server simply does not answer yet. It holds the request open until a message actually exists, then replies with it. This is called long-polling, and it genuinely delivers promptly — real chat products shipped on it for years. But look at what it does: it holds a connection open for the wait, then tears it down after every single message and immediately opens a new one. At millions of messages a second, that is millions of teardowns and re-opens a second, for connections we wanted open the whole time.

Both dead ends point at the same fact. If a server must be able to reach a phone at any moment, something has to stay open between them the whole time. Once you accept that, the sensible move is to hold that connection once, openly, and stop paying to recreate it.

Persistent WebSocket. Every online phone opens one WebSocket and keeps it open. A WebSocket starts life as an ordinary web request carrying an Upgrade header. After the server accepts it, the connection does not close — it stays open and becomes two-way, which means the server can send data down it at any time, without being asked. That is exactly the property polling could never buy. A message for you arrives at the server, the server writes it down your socket, and it is on your screen. No asking, no gap.

Two housekeeping details make this real. First, a heartbeat: a tiny ping and pong travels over each connection every ten to thirty seconds, so the server notices when a phone has silently vanished — a dead battery, a lost signal — and cleans up. Second, the server itself changes character. It is no longer answering quick requests and forgetting you; its whole job is now holding your connection open. That makes it a different kind of server, so we rename it for what it now does: a chat server. The trade-off we accept is that a chat server is stateful — it holds particular people's live connections, so losing one matters. What that costs, and how we survive it, is the last use case.

Under the hood — how it works

What is this connection, physically? A WebSocket runs on top of TCP. When two machines talk over TCP, the operating system on each end creates a socket — a small in-kernel object that represents that one conversation. Every TCP connection is identified by four numbers together: the source IP address, the source port, the destination IP address, and the destination port. This is called the 4-tuple. When a packet arrives, the operating system reads those four numbers off it and matches the packet to the right open socket. Change any one of the four, and it is a different connection.

A port is a sixteen-bit number, so it runs from zero to about sixty-five thousand. That one fact is the source of a famous myth: 'a server can only hold 65k connections.' Here is where the ceiling really applies. When your phone connects to one server, three of the four numbers are already fixed — the server's IP, the server's port (say 443), and the phone's own IP. Only the phone's source port can vary. So one client can hold at most about sixty-five thousand connections to one server address. That limit is real. But notice who it binds: a client, talking to a single address. No phone needs sixty-five thousand connections to one server.

Now flip to the server side, because this is the part people get wrong in interviews. On every connection a server accepts, the server's own two numbers are the same — its IP and its one listening port. What varies is the client side: billions of possible client IPs, each with its own range of ports. Two clients that happen to use the same source port are still different 4-tuples, because their IPs differ. So the server spends no port per client. It reuses its one listening port for everyone, and the sixty-five-thousand ceiling simply does not apply to accepting connections.

So what does limit a server? Each open connection costs a file descriptor — the small integer handle the operating system gives out per open socket — plus a few kilobytes of kernel memory for buffers. The default file-descriptor limit is low, often around a thousand, but it is just a setting, and you raise it. Give the box enough RAM, and one server holds millions of connections. WhatsApp famously tuned exactly these knobs and held over two million connections on a single machine. The real ceiling is millions, not tens of thousands.

One place the port ceiling does bite a server: outbound connections. The moment a server opens many connections to one other machine, it is the client in that relationship — and it faces the same sixty-five-thousand ceiling toward that single address. Keep that in your pocket. It comes back the moment our servers need to talk to each other in bulk.

What does a chat server actually do all day? It listens on a port. When a phone connects, the OS hands it a fresh socket, and the WebSocket handshake upgrades that socket into a persistent two-way channel. Then the server watches it — along with the half million other sockets it holds — for incoming data. It cannot afford a thread per connection; a million threads is not a thing an operating system will give you. Instead it uses the kernel's event-notification machinery (epoll on Linux, kqueue on BSD): one thread hands the kernel its whole list of sockets and asks, 'which of these have data right now?' The kernel returns just the handful that do. That trick is what lets one box watch a million connections with a few threads.

One small piece completes the picture: with half a million sockets open, how does the server find the right one for a given phone? It keeps an in-memory table mapping each connected device to its socket's file descriptor. On connect, it records the entry. To deliver, it looks up the device, gets the descriptor, and writes the message to exactly that socket. On disconnect, the entry is removed. Plain, per-server, in RAM.

Put together, this explains what actually caps a chat server. Holding idle sockets is cheap — descriptors and RAM, both of which just sit there. But our sockets are not idle. Every message means the kernel waking a thread, bytes being copied, the right connection being found and written to. With connections churning messages, the CPU runs out long before the socket count does. A chat server is CPU-bound, not socket-bound. Hold that thought for the next deep dive, where we size the fleet.

deep dive 2 The connection fleet
The question

Every online phone now holds one open connection, and those connections have to live somewhere. Of our billion users, roughly a fifth are online at any moment — two hundred million open connections. So before a single message moves: how many servers does it take just to hold them? That comes down to one number — how many connections we design each box to carry.

Start with the number everyone reaches for: about sixty-five thousand connections per server. It comes from the size of a port number, and it is a myth — that ceiling limits a client talking to one address, not a server accepting connections, because the server reuses its one listening port for every client. (The full story, from first principles, is under the hood in the last deep dive.) If you believed the myth, you would size a fleet of over three thousand boxes against a wall that is not there.

The real fork is between two believable numbers. WhatsApp proved that one tuned box can hold about two million connections. Design to that hero number and the fleet is only about a hundred boxes. Or design to something deliberately smaller — half a million per box — and run about four hundred.

The hero number has a catch: it is an idle ceiling. An open connection that is sitting quietly costs almost nothing — a handle and a few kilobytes of memory. Moving messages across it is what costs: every message wakes the server, gets copied through it, and gets written to exactly the right socket, and all of that burns CPU. Two million quiet connections fit on one box; two million busy ones do not. Pack a box to its idle ceiling and there is nothing left for the actual work.

Conservative — half a million per box. We design each chat server to hold five hundred thousand connections. That deliberately leaves most of the box's capacity free for processing messages, which is the load that really caps it. The arithmetic is then one line, and it is the line that defines chat at scale: two hundred million connections, at half a million per box, is about four hundred chat servers. Four hundred machines doing nothing but keeping sockets open, before one message is stored. We accept the bigger fleet on purpose — it is the price of headroom.

New connections need to be spread across that fleet, so a load balancer goes in front. When a phone connects, the balancer picks a chat server for it, and once the WebSocket is open, the connection stays on that server until it drops — 'sticky' just means the balancer places a reconnecting device predictably. What is the balancer itself? Unlike a chat server, it keeps no per-person state: it forwards a connection's bytes without reading them, which is such light work that a small set of balancer machines fronts the whole fleet, sharing one public address. Losing one balancer does drop the connections passing through it — but they reconnect through the survivors within seconds, and unlike a chat server, nothing about any person lived on the box.

Under the hood — how it works

If you designed to the two-million hero number instead, the same arithmetic gives about a hundred boxes. The point of this exercise is the shape and the order of magnitude — a fleet sized by connection count, not by message rate — more than the exact figure. Any defensible per-box number between those two gives a fleet in the hundreds.

One more word on what this fleet is not: the four hundred boxes are not four hundred copies of each other. Each one holds a different half million people's live sockets. That is what 'stateful' means here, and it is why losing one box is a real event — the last use case deals with exactly that.

phoneChat serverpersistent connsheartbeat · ×400500k/box · stickydatabaseLBsticky sessions
The board after this deep dive.
use case 2 Routing one message

Now two phones are online, and A sends B a message. This use case is the heart of the whole design, and it breaks into two problems: first A's server has to locate B, then it has to get the message across to whatever server B is on.

deep dive 3 Finding the recipient
The question

A's chat server is holding A's message. B's socket is on one of four hundred servers, and nothing so far says which. How does A's server find the one server that holds B — for every message, three million times a second?

The crudest answer is to not find B at all: shout. A's server sends the message to all four hundred servers; the one holding B delivers it, and the other three hundred ninety-nine ignore it. No lookup, no bookkeeping to maintain — and every 1:1 message now costs four hundred times the work. At three million messages a second, that is over a billion server-to-server touches a second. It is ruled out on the spot — but worth saying out loud, because 'avoid the broadcast' is the requirement every real answer has to meet.

A much better idea: make B's location computable instead of stored. When a phone connects, choose its server by hashing its id — running the id through a fixed function that turns it into, in effect, a server number. Then any server can re-run the hash on B's id and know where B is, with no lookup at all. This idea is half right, and we keep the half that works. We do place connections this way, because it spreads people evenly across the fleet. But as the source of truth it has a flaw: the hash tells you where B should be, not where B is. B's assigned server was full, so B landed elsewhere. B's server died, so B reconnected somewhere else. B opened the app on a laptop — a second device the hash knows nothing about. The hash goes stale exactly when reality moves, and in chat, reality moves constantly.

Third idea: let the servers tell each other. Each chat server gossips who it is holding, and eventually every server knows where everyone is. Gossip is genuinely the right tool for a small, slowly-changing fact — like which of the four hundred servers are alive right now, and the fleet does use it for that. But per-user location is two hundred million facts that change on every connect and disconnect. Word of mouth lags, and here a lag has a concrete cost: someone who just connected is not yet known everywhere, so their first messages miss.

Follow all three failures and they point at the same requirement: a single place that always knows, updated at the exact moment the truth changes — when a socket opens or closes.

A connection directory. One shared lookup table: for every online device, which chat server holds its socket right now. The moment your phone's connection opens, your chat server writes your entry. The moment it closes, the entry is cleared. To route a message, A's server asks one question — where is B? — gets one answer — chat server 340 — and forwards the message there. Chat server 340 finds B's socket in its own in-memory table and writes the message down it. One lookup, one hop, done.

What is this table, concretely? It is a plain key-value lookup at a very high read rate, and the standard tool for that job is Redis — an in-memory store where a small read costs well under a millisecond. Memcached, the other common in-memory store, could hold the same data — but Redis comes with clustering and replication built in (replication meaning live copies on other machines), and this table has to survive its own machines failing; the last use case leans on exactly that. Now size it, by traffic rather than by bytes. Every online delivery does one lookup, so the directory serves about three million reads a second at peak. One Redis machine comfortably handles about a hundred thousand small operations a second. So we split the table into about thirty shards — a shard being one slice of a store that has been split across machines — with the device id hashed to pick the shard. The bytes, by contrast, are almost nothing: two hundred million entries at about a hundred bytes each is roughly twenty gigabytes across the whole cluster. Thirty machines for the read rate, a trivial amount of data — the directory is sized by throughput, not capacity.

The cost, said plainly: this is one more component that every single delivery depends on. If the directory is slow, chat is slow; if it is down, routing is down. We take that deal because the alternatives are worse, and because the dependency can be hardened — which is exactly what the last use case does.

A's phone
1 · send
A's chat server
2 · where is B?chat server 340
directoryRedis · device→server
3 · forward
chat server 340
4 · push down B's socket
B's phone
One 1:1 message, end to end
key — devicevalue — chat server
B's phonechat server 340
B's laptopchat server 12
A's phonechat server 17
Under the hood — how it works

How placement works, and what consistent hashing does and does not do here. When a phone connects, the balancer picks its chat server by hashing the user id — that spreads people evenly, and it is only about placement. It never moves a live connection. Add a new chat server and nothing tells two hundred million open sockets to migrate: the hash ring changes for future placements only, and a device lands on the new server the next time it naturally reconnects — a dropped network, an app restart. At that moment its directory entry is rewritten to the new location. So the hash spreads load, and the directory tells the truth. Consistent hashing's specific virtue is that adding or removing a server reshuffles only about one server's share of future placements instead of everyone's — the same primitive the message store reaches for later when it splits across nodes.

Why the directory needs no heavy machinery: an entry is just 'who holds this socket right now,' and it is rewritten every time the device reconnects. If an entry is briefly wrong — say its server just died — the forward misses, the miss is noticed, and the message is handled as if B were offline, a case a later use case makes safe. A fact this cheap, that corrects itself on the next reconnect, does not need a consensus protocol. It needs to be fast.

phoneChat serverpersistent connsheartbeat · ×400500k/box · stickydatabaseLBsticky sessionsConnectiondirectory · Redisdevice→serversharded · replicated
The board after this deep dive.
deep dive 4 Server to server
The question

The routing answer said 'forwards the message to chat server 340' as if that were nothing. How, exactly? The obvious way is a direct network connection from A's server to B's server. For one pair of servers, that is fine. But any of our four hundred servers may need to reach any other at any moment. Should every chat server hold a direct line to every other — or should there be one shared thing in the middle?

  the mesh (every server wired         the broker (every server holds
  to every other)                      one connection)

    S1 ─── S2                             S1   S2
    │ ╲   ╱ │                              ╲   ╱
    │  ╳   │                              [broker]
    │ ╱   ╲ │                              ╱   ╲
    S3 ─── S4                             S3   S4

  400 servers ≈ 80,000 links            400 servers = 400 links
Direct mesh vs one shared broker

Work out what the direct mesh costs first, because for a small fleet it is genuinely the right call. With four hundred servers, each may need to talk to the other three hundred ninety-nine — about eighty thousand distinct links across the fleet. The sockets themselves are not the problem; we just saw one box hold half a million. The problem is the bookkeeping. Every server now tracks the health of every other server, reconnects to each one that drops, and rebalances every time the fleet grows. That workload grows with the square of the fleet size, and it lands on the exact tier whose CPU we are protecting. It works fine in the design review and turns into an operational swamp as the fleet grows.

So: one shared thing in the middle that every chat server talks to. Each server then holds one connection instead of three hundred ninety-nine, and nobody tracks anybody's health except the middle thing's. The pattern is called a message broker. That settles the shape — but not the product, and the products are not interchangeable.

Redis pub/sub is the lightest candidate, and Redis is already in our stack. But it is fire-and-forget: if a server is disconnected for even a moment, anything published to it in that gap is simply gone, and we would end up rebuilding delivery guarantees on top of it ourselves. RabbitMQ is a real broker with real guarantees, built around routing individual tasks to the programs that consume them — excellent up to tens of thousands of messages a second, which leaves it two orders of magnitude short of our three million. Kafka is different in kind: it treats messages as a stream you append to, and it is built to split that stream across machines, which is exactly the property our load demands. Kafka it is — and the answer below shows how it carries the load.

A shared message broker (Kafka). Every chat server holds exactly one connection — to Kafka. To forward a message, A's server publishes it, tagged for chat server 340. Chat server 340 is continuously reading its own stream, sees the message, and writes it down B's socket. All the all-to-all traffic still happens; only the wiring collapsed, from about eighty thousand managed links to four hundred — one per server.

How does Kafka itself carry three million messages a second? Partitions. A stream in Kafka is split into many partitions — independent slices, each living on some machine in the Kafka cluster — and we partition by the target chat server, so each chat server reads just its own slice. No single broker machine ever sees the whole flood, and adding chat servers just adds partitions, so the broker scales with the fleet. In machines: at about a kilobyte per message, three million a second is roughly three gigabytes flowing in and out per second, which a cluster of a few dozen broker machines carries comfortably. Each partition is also kept on disk for a short window, so a chat server that crashes and comes back resumes its stream from where it stopped — forwards that were in flight are not lost. And the broker must not become the new weak point: Kafka keeps copies of each partition on more than one machine in the cluster, so losing a broker machine loses nothing.

One thing the broker does not do: decide whether B is online. That is still the directory's one question. The broker only carries the forward once the directory has answered.

candidatewhat it isat 3M messages/s
Redis pub/subin-memory publish/subscribe, fire-and-forgetfast, but a briefly-disconnected server loses whatever was published in the gap
RabbitMQa broker that routes individual tasks to consumersgreat at moderate volume; this throughput is past what a Rabbit cluster carries happily
Kafkaan append-only stream, split across partitionsbuilt for this: partitions spread the load, a short durable window lets a crashed reader resume
partition 1msgmsgmsgread by chat server 1
partition 2msgmsgread by chat server 2
partition …one per chat server — add servers, add partitions
The topic, partitioned by target chat server
Under the hood — how it works

Both directions ride the same connection. A chat server is a producer when it forwards — publish, tagged with the target server — and a consumer of its own partition when it receives. There is no request and reply between servers at all; handing the message to the target's partition is the hand-off.

Remember the note from the first deep dive — that a server opening bulk connections to one address becomes a client, with a client's port ceiling? The broker quietly settles that too. Each server's outbound need is one connection to the Kafka cluster, not hundreds of connections toward whichever peer happens to be busy.

phoneChat serverpersistent connsheartbeat · ×400500k/box · stickydatabaseLBsticky sessionsConnectiondirectory · Redisdevice→serversharded · replicatedMessagebroker · Kafkapartitioned by target serverone conn per server
The board after this deep dive — “Send a message” traced, hop by hop. The numbered steps below walk the same path.
Send a message
  1. the phone reaches the load balancer over the public internettens of ms
  2. and lands on the chat server that holds its socket~2 ms
  3. the chat server stores the message durably and answers the sender — 'sent' now means 'kept'~10 ms
  4. the chat server looks the recipient up in the directory — device → chat server 340~5 ms
  5. it publishes the message onto the broker, tagged for chat server 340~2 ms
  6. chat server 340 reads its own stream, picks the message up, and pushes it down the recipient's socket~2 ms
use case 3 Keeping messages, in order

So far a message has traveled from A to B, but we never asked what holds it once it lands. Two of our promises live here: messages come back in order, and none is ever lost. Keeping both at this write rate takes three linked decisions — where messages are stored, how that store is split across machines, and what stamps each message with its order.

deep dive 5 The message store
The question

Time to look at the box we have been writing into. The day-one database has been quietly storing every message before its sender hears 'sent,' and it cannot keep doing it: a single node takes tens of thousands of writes a second, and peak is three million. So: what kind of store takes three million writes a second and still serves our one read — the recent messages in this chat — cheaply?

Look at how this system reads and writes, because the workload picks the store.

We only ever do one kind of read: open a chat and load its most recent messages. That is the whole read story. No joins, no searching across chats, no analytics on this path. The writes are the opposite story: constant, tiny, three million a second at peak, forever.

A relational database — Postgres, MySQL — is the default reach, and here it is the wrong fit twice over. It offers rich queries we would never run. And in exchange, a single node gives us tens of thousands of durable writes a second, when we need a hundred times that. Worse, a relational database does not spread one table across a hundred machines by itself — making it do that is a project you own forever. We would be paying for a query engine we never use while fighting its scaling from day one.

What about a log, like Kafka, since messages are already an ordered stream? A log is a superb transport — we just made it one. But as a store, it answers the wrong read. A log reads 'the whole stream, in order.' Our read is 'jump into one particular chat and give me its recent tail,' and you cannot jump into the middle of a log by chat cheaply.

There is a family of stores whose native shape is exactly our workload. It groups rows by a key and keeps them sorted inside each group. Writes are fast appends. Reading the recent tail of one group is cheap. And when you need more capacity, you add machines, and the store spreads itself across them — that part is built in. This family is called the wide-column store.

A wide-column store. Concretely: Cassandra, or its faster rewrite ScyllaDB — the store family Discord and many chat products run on. Messages are grouped by chat and kept in order inside each chat. Opening a conversation reads one group's recent tail. Writing is a fast append. Adding capacity is adding nodes. And the trade we sign is just as concrete: this store does the one read we designed for and nothing else. A question like 'every message this person sent last Tuesday, across all their chats' is not cheap here — if we ever need it, we build a separate index for it. We specialize, on purpose, because the workload is that lopsided.

candidateour read — a chat's recent tailour writes — 3M/s
relational (Postgres)finea single node tops out in the tens of thousands
a log (Kafka)can't jump to one chat cheaplyfine
wide-column (Cassandra / ScyllaDB)one group's tail — cheap by designfast appends, scale by adding nodes
chat c-92f#1#2#3#4read the tail →
chat c-7aa#1#2
chat c-e01#1#2#3
Messages grouped by chat, ordered within each group
Under the hood — how it works

Why these stores swallow writes so easily: they are LSM-based. A write lands in memory and in an append-only log at the same time — durable immediately, so a crash loses nothing — and gets merged into sorted files on disk later, in the background. Deferring the expensive disk work is the whole trick. 'The last messages in chat X' becomes 'read the top of chat X's group': the one read, kept cheap.

How the storage engine does all this on the inside — the memtables, the sorted files, the background compaction, how the cluster places and replicates each group of rows — is a rich topic of its own, and this design only needs the two facts above: writes are fast appends, and a chat's tail is one cheap read. The linked sim below shows the write path in motion, and the storage-engine internals will get their own full topic on this site.

⚡ From production

Discord runs this exact shape at trillions of messages — grouped by channel and time, first in Cassandra, later migrated to ScyllaDB when long garbage-collection pauses hurt their read latency. Their read pattern is the same as ours: 'the recent messages in this channel.' The lesson worth carrying: the store family was right at their scale and stayed right — what changed under load was the implementation, not the data model.

Discord — How Discord stores trillions of messages
deep dive 6 The partition key
The question

The store is chosen; now it has to be split — the same sharding move the directory made, this time on the big table. One node takes about fifty thousand writes a second and peak is three million, so the store becomes about sixty shards, each on its own machine. One column of the row decides which shard each message lands on. Look at the table: which column would you split by, so that writes spread across all sixty shards and reading one chat stays cheap?

table messages — the data under this deep dive
chat_iduuidwhich conversation it belongs to
sender_iduuidwho sent it
created_attimestampwhen it landed
seqbigintposition within the chat — payload
bodytextthe message — payload

Take the three candidate columns one at a time, and test each against both jobs: spread the writes, keep the read on one shard.

Split by created_at, the timestamp? It sounds sensible — recent messages are the hot ones. But watch what happens to writes. Every message being written right now carries the same 'now,' so every single write lands on whichever shard owns the current time slice. One shard takes the entire three million a second while fifty-nine sit cold. And the read gains nothing, because we look messages up by chat, not by time. This is the trap answer: a true observation attached to the wrong mechanism.

Split by sender_id? Now the writes spread nicely — there are hundreds of millions of senders. It is the read that breaks. A conversation's messages come from many senders, so one chat's history is scattered across every member's shard, and opening a chat — the one read we optimized the whole store for — becomes a gather across the fleet. Hold onto the idea, though: a per-person split is exactly right for a different table, coming in the offline use case.

Split by chat_id? Every message in a conversation lands on the same shard, so opening a chat reads one shard. And with billions of mostly-small chats, writes spread across all sixty on average. The one weakness is honest and visible: a giant, hyperactive group is still one chat on one shard.

chat_id. Split by the key you read by: the store shards about sixty ways on chat_id. The giant-group weak spot from the debate is real, and it is a known hot spot with a planned fix — described under the hood below — rather than a surprise.

One check before moving on: 'sixty shards' answers only one of two different sizing questions. Sixty is the throughput answer — three million writes a second, over fifty thousand per node. Capacity is a separate and bigger answer. At about a kilobyte per message and a hundred billion messages a day, the store grows by roughly a hundred terabytes every day — even one year of that is tens of petabytes, which is why real chat products either cap how much history they keep or move old messages to cheaper storage. So disk and retention, not write rate, set the true machine count. Each shard is also replicated — about three copies — so a dead machine loses nothing. Put together, the real fleet is a couple hundred machines: sixty is the write floor; capacity and replication set the rest.

Under the hood — how it works

The fix for the one hot chat: when a single chat outgrows its shard, spread that chat's writes over a handful of sub-buckets — the key becomes the chat id plus a small bucket number — and merge the buckets back together on read. The rows scatter; the conversation does not, because every message carries its position in the chat (the next deep dive's sequence number), and the read reassembles the buckets by that. You trade the clean one-shard read, for the handful of chats that need it, in exchange for write load the shard can survive.

⚡ From production

This is Discord's hot-partition story, exactly. They split messages by channel; a few huge, busy channels concentrated onto single shards, and read latency on those shards spiked while quiet ones idled. Their fix kept the key and changed the machinery around it — most notably merging many simultaneous reads of the same hot channel into a single read. The key was right: split by what you read by, then engineer for the tail.

Discord — How Discord stores trillions of messages
phoneChat serverpersistent connsheartbeat · ×400500k/box · stickyMessage store · Cassandra60 shards · by chat_idLBsticky sessionsConnectiondirectory · Redisdevice→serversharded · replicatedMessagebroker · Kafkapartitioned by target serverone conn per server
The board after this deep dive.
deep dive 7 Message ordering
The question

Two messages hit the same chat within the same millisecond. Which comes first? And when a phone reconnects after an hour away, how does it fetch exactly what it missed — no gaps, no duplicates? Both questions are really one question: what defines the order of messages in a chat?

The natural first answer is the timestamp — every message already carries one, so sort by it. But 'sort by time' quietly assumes two things that are false. It assumes no two messages share a time; at three million a second, two messages in the same chat land in the same millisecond all day long, and then there is a tie. And it assumes every server's clock agrees; they do not — real clocks drift apart by milliseconds. Neither error sounds dramatic, but our promise is that a reply never renders above the message it answers, and a few milliseconds of drift breaks exactly that.

The opposite extreme is a single global counter: every message in the whole system gets the next number, one perfect worldwide order. It works — and it costs everything, because every message on earth now queues at one counter for its number. That is a global bottleneck, bought for a guarantee nobody uses: no one ever reads all chats interleaved into one stream.

Between 'free but wrong' and 'perfect but ruinous' sits what the product actually promises: order matters within one chat. So put the coordination there, and only there.

Sortable id + per-chat sequence. Every message gets two numbers, and each does one job. The first is a sortable id — a plain sixty-four-bit number any server can mint on its own, with the time in its high bits, so ids sort roughly by time across the whole system with zero coordination. (The exact layout is in the figure below.) The second is the per-chat sequence: one, two, three within each conversation. This is the number that actually guarantees order.

So who hands out the sequence? Nothing new on the board. A chat's messages all land on one store shard — that was the point of splitting by chat — and each shard has a lead copy that accepts its writes. That lead copy is the chat's one writer: as it accepts each message (the same durable write that lets the sender hear 'sent'), it assigns the chat's next number from a counter in its memory. One writer per chat means the numbers come out gap-free, with no coordination beyond that one conversation — which is naturally one-at-a-time anyway. If that machine dies, another copy of the shard takes over, reads the highest sequence already stored for the chat, and continues from there. A lost counter costs one lookup, never a gap.

The sequence also answers the second question — catching up. 'Give me everything in this chat after the last sequence number I saw' is an exact request: no gaps, no duplicates, nothing to guess. That is the after_seq cursor in the catch-up API, and it is how a reconnecting phone fills in what it missed.

┌─────────────────────────────┬────────────┬─────────────┐
│ timestamp — 41 bits         │ machine id │ counter     │
│ (milliseconds; ~70 years)   │ 10 bits    │ 12 bits     │
└─────────────────────────────┴────────────┴─────────────┘
  high bits first → comparing two ids compares time first

  example, minted on machine 7, third id that millisecond:
  [ 1720950000123 ][ 0000000111 ][ 000000000011 ]
The sortable id — 64 bits, three fields
Under the hood — how it works

The sortable id's exact layout — it is the Snowflake scheme. Sixty-four bits, one eight-byte integer, in three fields. The top forty-one bits hold a millisecond timestamp (enough range for about seventy years). The next ten bits hold a machine id, so a thousand-odd servers can mint ids at the same time without ever colliding. The last twelve bits are a counter the machine bumps for each id it mints within one millisecond — about four thousand per machine per millisecond. To mint one: take the current millisecond, put it in the high bits, drop in your machine id and your counter. Because the timestamp sits on top, comparing two ids compares their times first — that is what makes them time-sortable with no coordination at all.

The single sequence owner is also the real ceiling, so name it. Sub-bucketing (from the last deep dive) spreads a hot chat's storage across machines — but every one of that chat's messages still visits its one sequence owner for a number. So a single white-hot chat's throughput is bounded by that one owner. For chat this is fine: one conversation's messages are serial by nature, and no phone renders faster than that. But the limit is the owner, not the disks — a senior answer says so out loud.

use case 4 Delivering to someone offline

Everything so far assumed B was online. Now B's phone is off the network entirely — no signal, or simply switched off — and yet A's message still cannot be dropped on the floor. So what happens to a message with nowhere to go yet?

deep dive 8 Offline delivery
The question

The directory lookup just came back empty: B has no live socket. The message still cannot be lost. So where does it wait until B's phone wakes — and how does the phone learn that it is waiting?

The tempting shortcut is the platform push service itself. Apple's APNs and Google's FCM exist to reach sleeping phones — so why not just send the message text through them and be done? Because they are someone else's system. They rate-limit you, they may merge several notifications into one or quietly drop some under pressure, and they hold none of our promises: no ordering, no durability, no catch-up. A push that carries the message is a message whose delivery you can no longer vouch for. So the push gets a smaller job: waking the phone. It is good at that.

The other shortcut fails faster: leave the message parked on a chat server until B comes back. But a chat server holds live sockets and it can die — and anything parked on it dies too. We already have a place whose whole job is keeping messages safe; the chat-server tier is not it.

So the job splits in two. Holding the message durably is our side: a mailbox in the store. Waking the phone is the platform's side: a push with nothing inside it.

Offline inbox + push. Two pieces, each doing the half it is good at. A durable offline inbox: a per-recipient mailbox that A's chat server appends the message to, so it survives — for hours or for weeks — until B returns. And a push notification, handed to the platform's service (APNs on iOS, FCM on Android), that carries no message content at all. It only tells the sleeping phone 'you have messages,' so the phone knows to reconnect and pull them from us.

Step by step, the same way we walked the send: (1) A's chat server looks B up in the directory and gets 'no live socket.' (2) It appends the message to B's inbox — a durable write, so nothing can be lost from here. (3) It hands a wake-up to APNs or FCM for B's device. (4) B's phone wakes, opens its WebSocket to a chat server, and asks for its inbox. (5) The chat server reads B's inbox in arrival order and streams the messages down the socket — each one carrying its per-chat sequence, so the phone drops it into the right conversation at the right position. (6) B's phone acknowledges what it received, and those inbox entries are cleared. From there, B is back on the live path.

What store is the inbox, concretely? No new technology: it is a second table on the same Cassandra-class store as messages, just keyed differently. Messages are keyed by chat; the inbox is keyed by device — whose mailbox it is. Each row carries the chat id, the message's per-chat sequence, and the body. Reading a mailbox is then one shard scan — every row for this device, in arrival order — exactly the shape of read this store is good at. Its write load is real, and the store's sizing already speaks its language: a message to offline recipients costs one inbox append per absent device, and you scale for those appends the same way as the messages table — by adding machines. The other costs we accept: one more table, and a dependency on Apple's and Google's push services for the wake-up. A lost push costs only latency, never the message — the phone also checks its inbox every time the app opens.

directoryanswers: no live socket
1
offline inboxappend — durable from here
2
push · APNs/FCMcarries no content — just 'you have messages'
3
B's phonewakes, reconnects, asks for its inbox
4 · drain, then clear
B's chat serverstreams the mailbox down in arrival order
Delivering to a phone with no socket
Under the hood — how it works

Two tables, and it matters which answers what. The inbox is keyed by device and answers 'what arrived for this phone while it was gone' — the wake-from-offline queue, drained and cleared on reconnect. The message store is keyed by chat and answers 'the messages in this conversation after a given sequence.' A phone that was only briefly gone — a dropped socket, a quick reconnect — skips the inbox entirely: it reconnects and catches up straight from the message store using its per-chat cursor — a bookmark of the last message it saw. Long-offline uses the inbox and a push; brief-offline uses the cursor. Same durability either way; two paths sized to two situations.

Why the inbox is keyed by device rather than person: one person can be on a phone and a laptop at once, and each device drains its own mailbox at its own pace — a message the laptop already showed still has to reach the phone. So the mailbox's one read is 'everything waiting for this device,' and the device is the key you read by.

phoneChat serverpersistent connsheartbeat · ×400500k/box · stickyMessage store · Cassandra60 shards · by chat_idLBsticky sessionsConnectiondirectory · Redisdevice→serversharded · replicatedMessagebroker · Kafkapartitioned by target serverone conn per serverOffline inboxdurable · keyed by devicedrain in arrival orderPush · APNs/FCMwakes the phone only
The board after this deep dive — “Deliver to an offline phone” traced, hop by hop. The numbered steps below walk the same path.
Deliver to an offline phone
  1. A's chat server looks B up — the directory answers 'no live socket'~5 ms
  2. it appends the message to B's durable inbox — from here, nothing can be lost~10 ms
  3. it hands APNs/FCM a content-free wake-up: 'you have messages'~5 ms
  4. the push wakes B's phone on the platform's schedule, not ours; the phone reconnects and drains its inbox in arrival orderseconds
use case 5 Messaging a big group

One recipient was easy. Now A sends to a thousand-member group, and the naive send makes A wait for all thousand handoffs before it hears a thing. A real app will not sit through that. So this use case pulls the slow part off the sender's path.

deep dive 9 Fan-out to a big group
The question

This deep dive opens with the design already broken. A sent to a thousand-member group; the send path is trying to deliver to all thousand members before it answers A, and A's app is timing out. What change gets A a fast 'sent' while a thousand deliveries still happen?

Name the two promises hiding inside the word 'send,' because they cost wildly different amounts. Promise one: your message is safe. That is what A is actually waiting to hear, and it costs one durable write. Promise two: everyone has it. That costs up to a thousand deliveries. The naive path welds the two together, so A waits on the expensive one.

Could we just add more chat servers? This is the trap answer, because it scales the symptom. Twice the servers still run the same thousand-delivery loop inside every group send — the sender still waits on the biggest group, just on a newer machine. The problem is not capacity. It is that the thousand deliveries sit on the send path at all.

Could we keep delivering inline, for correctness? For a five-person chat — honestly, yes. Many systems fan out small groups inline and only queue the big ones. It just cannot survive the tail, and the tail is in our requirements: a thousand members.

So separate the promises. Acknowledge A the moment the message is stored. Move the thousand deliveries off A's path entirely, onto something that can chew through them at its own pace.

Async workers off the broker. The sender's chat server does two cheap things and then answers. It stores the message — that is the promise A is actually waiting on. And it drops one small job onto the Kafka broker we already have: 'deliver this to group G.' Then it tells A 'sent.' Total cost on A's path: one durable write and one hand-off to the broker. Behind that answer, a pool of fan-out workers pulls the job, reads group G's member list from a small group-membership table (each group's list of member ids — read once per group send), and delivers to each member at its own pace: online members through the broker to their chat servers, offline members into their inboxes with a push. Nothing here is a new invention — it is the broker from the routing use case, plus a worker pool sized by delivery rate: scale it like any pool of interchangeable workers, by adding more when the backlog grows.

The trade this buys is worth saying plainly: group delivery becomes eventually consistent. A is told 'sent' while the last members have not received it yet, so for a moment some of the group has the message and some does not. For chat, that is the right trade — a small delivery lag inside a big group is invisible, and a multi-second freeze on every group send is not.

  A ── send ──▶ chat server ── 1 · store ──▶ message store
                    │
                    ├── 2 · one job on the broker: 'deliver to group G'
                    │
                    └── 3 · ack A  ✓ (~200 ms — A is done here)

  broker ──▶ fan-out workers ── read members of G (1,000) ──┐
                                                           │
                online member  ──▶ broker ──▶ their chat server ──▶ socket
                offline member ──▶ inbox + push
One group send: ack fast, deliver behind it
Under the hood — how it works

The failure story, because a worker pool earns its keep there. The job on the broker is durable, so a worker that crashes mid-fan-out loses nothing — the job is redelivered and picked up by another worker. Redelivery has one wrinkle worth naming: members already delivered to may get the message twice, and the message's id is what lets a phone recognize and drop the duplicate. If the workers as a group fall behind a burst, the backlog is visible on the broker and the fix is boring: add workers. The burst gets absorbed and drained, never dropped.

What is genuinely new here is small: the workers, and a group-membership store — a table mapping each group to its member list, read once per group send to expand 'group G' into its thousand recipients. From there each delivery is an ordinary 1:1 send made by the worker: ask the connection directory which chat server holds that member's socket (a thousand lookups, batched into a few round trips), then publish the message on that server's broker partition — or, if the directory has no entry, take the offline path: inbox plus push. The broker is not new. It was added to kill the server-to-server mesh, it already carries every 1:1 forward, and a group job is just a second kind of message on the same Kafka. That is the payoff of building the broker where the mesh was decided: the big-group case costs a worker pool, not a new piece of core infrastructure.

phoneChat serverpersistent connsheartbeat · ×400500k/box · stickyMessage store · Cassandra60 shards · by chat_idLBsticky sessionsConnectiondirectory · Redisdevice→serversharded · replicatedMessagebroker · Kafkapartitioned by target serverone conn per serverOffline inboxdurable · keyed by devicedrain in arrival orderPush · APNs/FCMwakes the phone onlyFan-out workers×N · expand + deliverGroup membershipgroup→members
The board after this deep dive — “Send to a big group” traced, hop by hop. The numbered steps below walk the same path.
Send to a big group
  1. A's chat server stores the message durably — the promise A is actually waiting on~10 ms
  2. it drops one small job on the broker — 'deliver this to group G' — and answers A 'sent'; A's part is over here~2 ms
  3. a fan-out worker picks the job up at its own pace~2 ms
  4. it reads the group's member list — one read that expands 'group G' into a thousand recipients~5 ms
  5. for each member it asks the directory which chat server holds their socket — the same lookup a 1:1 send makes, batched here~2 ms
  6. each online member gets the message through the broker — published on their chat server's partition — and down their socket~2 ms
  7. each member the directory has no entry for is offline and gets an inbox append instead —~10 ms
  8. — plus a push to wake their phone, exactly the offline path above~5 ms
use case 6 Delivery and read receipts

After you send a message, a little tick under it moves from sent to delivered to read. Each of those ticks is a fact that has to travel backward, from the reader's phone to the writer's. This use case works out what carries that fact, and why a big group makes it dangerous.

deep dive 10 Delivery & read receipts
The question

When B's phone receives A's message, A's bubble should tick to 'delivered'; when B opens the chat, to 'read.' How do those ticks travel back to A — and what stops a thousand-member group from burying A under a thousand of them?

Strip the mystery first: a receipt is itself a message. 'That message you sent has been read' is a tiny piece of text addressed to A — and we have just spent five use cases building a machine whose whole job is getting a tiny piece of text to a person, wherever they are, online or not. Building a separate receipts subsystem would mean building that machine twice. So receipts ride the machine we have, pointed the other way.

The one genuinely new problem is volume, and it comes from groups. Send one message to a thousand-member group and up to a thousand 'delivered' receipts and a thousand 'read' receipts come back — a thousand times as many acknowledgements as the message they describe. Sending each one back as its own message is perfectly right for a 1:1 chat — one tick each way — and unsurvivable for a big group. So the group path gets one extra move: batch the flood before it reaches the sender.

Send receipts back like messages, batched for groups. For a 1:1 chat, a receipt simply rides the machine backward. When B's phone receives or reads A's message, B's chat server sends a small receipt addressed to A — and it routes exactly like any message: look A up in the directory, forward through the broker, push down A's socket, and A's bubble ticks. If A is offline, the receipt waits in A's inbox like anything else. For a group, one extra move: the receipts from a thousand members flow back through the fan-out workers, and the workers batch them — one 'delivered to thirty people' update to A's phone instead of thirty separate messages.

One product question settles the storage: does a group chat show a count ('seen by twelve') or the list of who ('seen by Alice, Bob…')? Both travel the same way; they differ only in what you keep. A count is one number you bump as receipts arrive. The list keeps one row per reader — more storage, but you can show names. Pick per product; this design defaults to ticks with batched group counts.

1,000 members' phoneseach emits one small 'read' receipt
up their sockets
their chat servers
publish, tagged for the workers
message brokerthe same Kafka — a receipt is just a small message
fan in
fan-out workersthe same pool that delivered the message — bump a per-message tally
one batched update
A's chat serverreached back through the same broker
push down A's socket
A's phoneone update: 'read by 30 → 200 → 1,000'
A group's read receipts, batched on the way back
Under the hood — how it works

The group path in slow motion. When a member's phone reports 'delivered' or 'read,' its chat server publishes that receipt onto the same Kafka broker every message travels, just tagged for the fan-out workers instead of for a recipient's chat server. The same worker pool that fanned the message out collects the receipts fanning back in: instead of forwarding each 'delivered' to A, a worker bumps a per-message tally and periodically sends A one update — 'delivered to thirty,' then 'delivered to two hundred' — which travels to A like any other small message: the worker looks up A's chat server in the connection directory, publishes on that server's broker partition, and A's chat server sends it down A's socket. So the flood becomes a trickle before it touches A's socket, and it costs no new machinery: the same broker, the same workers, run backward.

Two honest footnotes. Batched counts are eventually consistent — 'delivered to thirty' may lag the true number by a moment, which is fine, because receipts are ambient status, not the message. And read receipts are a product and privacy choice: some users turn them off, and the machinery must respect that at the source — a phone with receipts off simply never emits the read receipt.

phoneChat serverpersistent connsheartbeat · ×400500k/box · stickyMessage store · Cassandra60 shards · by chat_idLBsticky sessionsConnectiondirectory · Redisdevice→serversharded · replicatedMessagebroker · Kafkapartitioned by target serverone conn per serverOffline inboxdurable · keyed by devicedrain in arrival orderPush · APNs/FCMwakes the phone onlyFan-out workers×N · batch receiptsGroup membershipgroup→members
The board after this deep dive — “A group's receipts travel back” traced, hop by hop. The numbered steps below walk the same path.
A group's receipts travel back
  1. each member's 'delivered'/'read' rides the same broker back, tagged for the fan-out workers~2 ms
  2. the same workers collect them and bump one per-message tally~2 ms
  3. when a batch is due, the worker asks the directory which chat server holds A's socket — a receipt routes exactly like a message~1 ms
  4. A gets one batched update — 'delivered to thirty' — published on that server's partition and down A's socket~2 ms
use case 7 Surviving a server dying

The whole system works now. This last use case keeps it working while its own machines die — and with four hundred chat servers, one of them is always down somewhere. The design has to survive that as a routine event, not an emergency.

deep dive 11 Surviving a chat server dying
The question

A chat server is stateful: it holds half a million specific people's live sockets, and those people's entries in the directory point at it. So when one dies, two things happen at once — half a million connections drop, and half a million phones notice within seconds of each other. What keeps that from becoming an outage?

Start with what actually happens in the second after a chat server dies, because the obvious safety argument misses it. The argument says: we have four hundred of them, losing one is a quarter of a percent, the fleet absorbs it. That works for stateless servers, where any survivor can serve any request. Our tier is not that. The dead box held half a million specific people's sockets, and no survivor has them.

So all half a million phones notice their dead connection within seconds of each other — and reconnect at the same moment. That reconnect stampede is the real danger: a wall of simultaneous reconnections landing on the balancer, the surviving servers, and the directory all at once, hard enough to knock over machines that were perfectly healthy. And a second problem hides behind the first: until each phone reconnects, its directory entry still points at the dead machine, so messages for those users are being forwarded to a box that cannot deliver them.

Redundancy did nothing about either problem. What survives the death is making the reconnection itself survivable, and making the one thing every delivery depends on — the directory — able to lose its own machines.

Replicate the directory + spread the reconnects out. Two moves, one for each half of the problem. First, the directory must have no single point of failure, because every delivery leans on it. It is already split into about thirty shards; now each shard gets replicas — copies on other machines — so losing one directory machine costs only a moment for one shard's slice, and even that slice heals itself as its users reconnect and rewrite their entries. Second, spread the reconnections out in time. Instead of every dropped phone retrying the instant it notices, each phone waits a short random delay first — so half a million reconnections land smeared across several seconds instead of in one spike. That deliberate randomness has a name: jitter. Together, the two moves turn a chat server death into a brief blip: dropped people are offline for a second or two, reconnect on healthy servers, and the directory follows them there.

  no jitter — everyone retries the instant they notice:

  load │█
       │█
       │█▁▁▁▁▁▁▁▁▁▁▁
       └────────────── time

  jittered — each phone waits a short random delay first:

  load │
       │▂▄▅▅▄▃▂▁
       │▁▁▁▁▁▁▁▁▁▁▁▁▁
       └────────────── time (a few seconds)
500k phones reconnecting: one spike vs jitter
Under the hood — how it works

Why the directory gets replication but not heavyweight consensus: a directory entry is cheap and self-correcting. It says 'who holds this socket right now,' and it is rewritten every time the device reconnects. If a replica takes over with a slightly stale entry, the cost is one missed forward, and the message takes the inbox path instead — never lost. Facts this cheap need speed and copies, not a consensus protocol. (The linked sim shows how a shard's replica takes over when its primary dies.)

The reconnect walk, end to end: phones detect the dead heartbeat; each waits its own short random delay; the balancer places them on healthy chat servers as they trickle in; each new connection rewrites its directory entry. Within seconds the dead box's half million people are spread across the fleet and the directory is current again. The people affected were genuinely offline for a second or two — for chat, invisible.

And the forwards that were in flight when the box died? Not lost. The dead server's slice of the broker sits untouched until a replacement picks it up — and it does not matter, because every one of those messages is already durable in the store. A phone that reconnects elsewhere catches up by its per-chat cursor and gets exactly what it missed. The forward was never the only copy; it was just the fast path.

⚡ From production

Slack hit exactly this. When one of their servers holding client connections drops, on the order of a hundred thousand clients reconnect at once — and each reconnecting client re-fetches its state, stampeding the backend. Their fix was a cache called Flannel running close to users, which serves that reconnect-time state so the herd never reaches the core — plus the same jittered, spread-out retries on the clients. The reconnect storm is a named, real failure mode; if reconnect state is expensive to compute, cache it near the clients.

Slack Engineering — Flannel: an application-level edge cache
Where this design sits on consistency (and CAP)

Sending chooses availability. If the sender's chat server can store the message, the sender hears 'sent' — even when some recipients are unreachable right now. The message is never refused; it waits in an inbox until its recipient returns. We chose that because, for chat, a send that fails outright feels worse than a send that arrives a little later.

Ordering chooses consistency, within one chat. A conversation's messages have one gap-free order, so replies always appear after the message they answer. And the scope stays deliberately small — one chat, never all chats at once.

Group delivery is eventually consistent. The fan-out runs behind the sender's ack, so for a moment some members of a big group have the message and others do not. We accept that because the alternative — deliver to everyone first, ack after — holds the sender hostage to the largest group.

There is a name for the underlying choice. When a network fault splits a system in two, each operation has exactly two honest options: answer anyway from what this half knows (stay available, risk being stale), or refuse until the halves agree (stay consistent, become unavailable). That either-or is the CAP theorem, and the useful way to apply it is per operation, by asking what a wrong answer would cost. Read this design that way: sends are available, per-chat ordering is consistent, group delivery is eventually consistent — each trade picked on purpose.

CAP theorem — the partition trade-off, per operation
The finished board

That is the whole machine — every box on it was placed by one of the deep dives above. If you carry away one sentence, make it the reframe we started with: chat is a routing system. Everything else on the board exists to keep that routing fast while the two promises — in order, never lost — stay true.

phoneChat serverpersistent connsheartbeat · ×400500k/box · stickyMessage store · Cassandra60 shards · by chat_idLBsticky sessionsstaggered reconnectConnectiondirectory · Redisdevice→serversharded · replicatedMessagebroker · Kafkapartitioned by target serverone conn per serverOffline inboxdurable · keyed by devicedrain in arrival orderPush · APNs/FCMwakes the phone onlyFan-out workers×N · batch receiptsGroup membershipgroup→members
The finished design — every box placed by one of the deep dives above.
What's on the diagram
phone
Your phone. It holds one live WebSocket to a chat server while you are online, and receives pushes when you are not.
Chat server
Chat servers. About four hundred boxes, each holding half a million people's live sockets and moving their messages.
Message store · Cassandra
Message store. A Cassandra-class wide-column store, sharded by chat — every message is written here before the sender is told 'sent.'
LB
Load balancer. Places each new connection on a chat server; once the socket is open, it stays where it was placed.
Connection directory · Redis
Connection directory. A sharded Redis cluster mapping each online device to the chat server holding its socket — written on connect, cleared on close.
Message broker · Kafka
Message broker. Kafka, partitioned by target chat server — every server-to-server forward and every fan-out job travels through it.
Offline inbox
Offline inbox. A per-device mailbox on the message store — where a message waits, durably and in order, when its recipient has no socket.
Push · APNs/FCM
Push (APNs/FCM). The platform's wake-up service. It carries no message content — it only tells a sleeping phone to reconnect and pull.
Fan-out workers
Fan-out workers. Pull group jobs off the broker, expand the member list, and deliver to each member — batching the receipts that flow back.
Group membership
Group membership. The table mapping each group to its member list, read once per group send.
What this design never covered — raise it yourself
Presence / online status

Seeing which of your contacts are online right now, and their 'last seen,' is a separate feature from holding your own connection. It comes up because the directory already knows who is online — and the temptation is to broadcast every connect and disconnect to everyone. The strong answer: presence is a fan-out problem, so when you come online, push the change only to the handful of people who are looking at you right now — the ones with a chat with you open on screen. Everyone else is sent nothing: their app simply reads your current status from the directory whenever they next open your chat. Do not fan every presence flip out to every contact — that is more traffic than the messages. (Typing indicators are the same mechanism, left out of scope.)

Multiple devices

One person runs the app on a phone and a laptop, and a message to them must appear on both — and a message they send from the phone must show up in their own laptop's copy. This is why the directory and the inbox key on the device rather than the person: each device holds its own socket and its own catch-up cursor. The strong answer: fan a message out to all of a person's connected devices, and give each device its own cursor so it catches up independently.

Media and attachments

People send a ten-megabyte video, and you must not push that through the message store and the broker. The strong answer: upload the media straight to an object store (S3 class) via a pre-signed URL — a temporary permissioned link the client uploads to directly — and put only the link (plus a thumbnail) in the message. The message stays about a kilobyte; the video's bytes go around the chat system, not through it.

Rate limiting and abuse

A single account or bot can try to blast messages far faster than a human — a spam and denial-of-service vector against the fan-out path. The strong answer: a per-user rate limit at the chat server (a token bucket — each user gets a steady refill of send allowance and is throttled when they exceed it), sized well above human speed but far below what a bot needs to do damage.

Go deeper — the primitives this design leaned on
  • TCP / WebSocketestablishes and keeps the persistent connection a chat server holds
  • Load balancerspreads new sticky connections across the fleet of chat servers
  • Consistent hashingplaces a connection and moves only one chat server's share when the fleet changes
  • LSM treetakes millions of writes a second by appending to memory and a log first, merging to disk later
  • Write-ahead logmakes the offline inbox durable the instant a message lands
  • Message queuethe broker between chat servers that also absorbs the fan-out burst
  • Rafthow one replicated directory shard elects a new primary when its node dies

Feedback on this problem →