1.16 Million Writes a Second: Sharding a Chat App’s Messages (System Design Interview in 30 Days, Day 5)

Why read replicas cannot carry write load, how to pick a shard key from the main query, and what gets harder once the data is split.

The chat app we sized on Day 2 stores 100 billion messages a day. Divide that by the 86,400 seconds in a day and you get about 1.16 million new messages every second, around the clock. Every one of them is a write. Today’s question is where those writes go, and the honest answer is that no single database server takes them. They have to be split across many servers. That split is called sharding, and it is the subject of Day 5 of the series.

Answering Day 4’s practice question

Yesterday ended with this: the messaging app writes about 1.16 million messages a second. Would read replicas help, and if not, what would?

Read replicas do not help with writes. A read replica is a copy of the database that answers reads. Every write still goes to the one primary first. The primary then streams the change to each replica, and each replica applies the same write again. So with two replicas, the cluster does the same write three times, and the primary still carries the whole load. Replicas add read capacity. They never add write capacity.

So the writes themselves have to be split. Each server takes only some of the conversations, and with them only that part of the writes. Each part is called a shard.

What sharding is

Sharding is horizontal partitioning where each partition lives on its own database server, with its own CPU, memory and disk. Partitioning a table on one server keeps the table manageable, but every partition still shares the same machine. Sharding is the step that adds write capacity, because each shard is an independent server.

A sharded system needs a routing layer that knows which shard holds a given row. That can be the application itself, a proxy such as Vitess, or the database’s own router, as in MongoDB or CockroachDB.

Picking the shard key from the main query

The shard key decides which shard a row lives on, and it is the most important decision in the design. The rule of thumb from the lesson: the shard key should be the column in the WHERE clause of 90% or more of your queries.

Get Roni Das’s stories in your inbox

Join Medium for free to get updates from this writer.

Remember me for faster sign in

For the chat app the main query is easy to name. The conversation screen asks for the last 50 messages of one conversation. So the key is conversation_id. All of one conversation's messages then live on one shard, and the query that runs every time someone opens a chat reads exactly one server.

The arithmetic, with the chat lesson’s numbers

The chat system lesson stores messages in Cassandra, partitioned by conversation_id, on a cluster of 50 to 100 nodes with a replication factor of 3, which means 3 copies of every message.

• 1.16 million writes a second times 3 copies is about 3.5 million writes a second across the cluster.
• Spread over 100 nodes, that is about 35,000 writes a second per node.
• Spread over 50 nodes, about 70,000.

That is the kind of number to say out loud in an interview. It shows why one machine is out of the question, and it gives the interviewer something concrete to push on.

What makes a good shard key

1. Many distinct values. user_id or conversation_id, not a yes or no column. A boolean gives you 2 shards.
2. An even spread. No single value should hold a large share of the rows. If one customer owns 40% of the data, their shard is a hotspot.
3. Most queries answered by one shard. If every query needs every shard, sharding made things slower.
4. A value that never changes. If the key changes, the row has to move to another server across the network.

Hash or range? Hash sharding computes hash(key) % number_of_shards. Rows spread evenly, but a range query such as "this week's messages" has to ask every shard. Range sharding puts users 1 to 1 million on shard 1, the next million on shard 2, and so on. Range queries stay on one shard, but if IDs grow over time, new rows pile onto the last shard.

What gets hard once you shard

• Queries without the shard key. The router sends them to every shard and merges the results. This is called scatter-gather, and its latency is the slowest shard plus the merge, not the average.
• Hot keys. One huge key, such as a giant group chat or a celebrity account, sends all its traffic to one shard. Adding shards does not help that one key. Fixes include salting the key across several shards, giving the hot tenant its own shard, or a cache in front.
• Changes that touch two shards. Moving credit from a user on shard 1 to a user on shard 3 is no longer one local transaction. Two-phase commit can block if the coordinator dies at the wrong moment, so most teams either keep such changes on one shard by design, or use a saga with compensating steps.

The usual mistake

Sharding first. The lesson lays out the order: tune the single server, add read replicas, partition big tables, split different areas into separate databases, and only then shard, when one server cannot take the writes for one area. Most applications that think they need sharding need better queries or caching.

Practice for Day 6

The messages sit on 4 shards, picked by hash(conversation_id) % 4. You add a fifth shard and switch to % 5. Work out roughly what share of conversations now belong on a different shard, and sketch how you would add shards without moving most of the data. We work through it at the start of Day 6.

Where this comes from

The sharding numbers and rules come from the database sharding lesson linked as the source, and the chat arithmetic from the chat system capstone. System Design Interview in 30 Days runs next to AI Engineering in 30 Days, which today covers serving a model behind an API. Both courses, System Design and AI Engineering, are on systemdesign.academy. You can read the sharding lesson itself here.