Replication, sharding and CAP
In short: Replication keeps copies of the data on several servers, sharding splits the data between them, and the CAP theorem says what you give up when the network splits.
Problem
One server for the whole shop
The shop grows. On Friday evening the catalogue is opened 20 000 times a minute, and almost all of it is reading: products, prices, stock. The database is on one server, and its processor is at 100%. In the night before Saturday its disk fails, and the shop is down until the morning.
One machine is both the ceiling of the speed and the single point where everything can stop. You need more machines with copies of the data.
The idea
Leader and followers
Replication means keeping the same data on several servers (nodes). The most common scheme has a leader (in PostgreSQL the primary): only it takes writes. The other nodes are followers (replicas). They get every write from the leader and apply it on their side. A read can go to any node, so three nodes take three times as many reads.
The box is a simulator of a cluster of three nodes, and the leader is A. The stock of white cheese (product 7) is 18 and is on all three. Time in the simulator stands still: a write leaves the leader and reaches the followers on the next tick. Every value has a version (v1, v2), so you can see which copy is newer. Under the output is the picture: the nodes, their copies and what is still in flight.
write on A returns OK at once: the leader has written on its side and waits for nobody. This is asynchronous replication. While the write travels, B still says 18. This delay is the replication lag. It is usually milliseconds, and under load it reaches seconds.
More precisely, if you are curious
The commands write, read, tick and the others in this lesson belong to the simulator, not to a particular database. tick moves its clock by one tick, while in a real database replication runs all the time. PostgreSQL does the same with streaming replication: the leader sends the followers its log of changes (the WAL), and they apply it in order. MySQL sends its binary log. In both, replication is asynchronous by default, and the followers only take reads.
Try it
I changed my city and the page shows the old one
Simona (customer 17) moves from Ruse to Plovdiv. She saves the new city in her profile, and the page that opens next still says Ruse:
The write went through, but the page read from B, and B has not got it yet. Other customers will see the new city a moment later, and for them it makes no difference. To Simona it looks as if the change was lost, and she types it again. What is needed here is the guarantee read-your-writes: whoever wrote sees their own write. The usual fix is that for a while after a change (a minute, say) the reads of that customer go to the leader. You saw the same out-of-date copy with the cache in the key-value lesson.
Save Simona's new city and read it right away so that she sees Plovdiv.
Solution
write customer:17:city = Plovdiv on A
read customer:17:city from A
read customer:17:city from A: the leader has every confirmed write. A tick and then a read from B also shows Plovdiv, but in a real database you do not know how long to wait. A read from B or C right after the write gives Ruse.
The idea
Synchronous replication and its price
If the leader goes down before the write reaches the followers, the write is only on the leader. So you can make it wait. With synchronous replication the leader says OK only when the followers confirm that they have the write. Then the confirmed write is on every node.
The price is twofold. Every write waits for the slowest follower, and between two data centres that means milliseconds on every write. And one dead follower stops all writes, because the leader has nobody to get a confirmation from. So in practice the two are mixed: one follower is synchronous, the others asynchronous. If the synchronous one goes down, another takes its place.
More precisely, if you are curious
In PostgreSQL the synchronous followers are set with synchronous_standby_names. FIRST 1 (b, c) waits for the first available one from the list, and ANY 1 (b, c) waits for either of the two. If nobody can confirm, the transaction hangs until a follower appears or the setting changes. The simulator does not hang: it refuses the write at once and says why. MySQL has semi-synchronous replication: the leader waits for at least one follower to confirm that it has received the write.
See it
The leader goes down
Replication is asynchronous. A customer buys white cheese, A lowers the stock to 17 and says OK. A moment later A stops, before the write leaves:
B and C are followers and do not take writes, so the shop is stuck until there is a new leader. elect picks, among the nodes that can see each other, the one with the newest data. It needs at least two of the three, a majority. Every election has a number (the term). The new leader has term 2, and when the old one comes back it learns that it is no longer the leader. Changing the leader is called failover.
But B never got 17. The confirmed sale vanishes: A drops it when it comes back, because the new leader does not have it. This is the trap of asynchronous replication: on failover the last confirmed writes can be lost. With synchronous replication B would have had it.
More precisely, if you are curious
MongoDB changes the leader (the primary) by itself: if the followers do not hear it for 10 seconds (the default), they vote, and a node with a majority of the votes wins. The writes of the old primary that did not reach the majority are rolled back when it rejoins. That is why important writes are sent with writeConcern: { w: "majority" }: they are confirmed only once the majority has them. PostgreSQL does not change the leader by itself. External tools such as Patroni do it, and by hand a follower becomes the leader with pg_promote(). In the simulator the election is manual too (elect).
The idea
The network splits: CAP and quorums
A dead server can at least be recognised. It is trickier when everything runs but the nodes cannot see each other, for example when the link between two data centres breaks. This is a network partition. The node on the other side does not know whether the others are down or it just cannot hear them.
A different scheme helps here, one without a leader. Every node takes writes and sends them to the others. A write waits for W confirmations, and a read asks R nodes and takes the newest version. If there are N nodes and W + R > N, every read asks at least one node that has the last write. This is a quorum. With N = 3, W = 2 and R = 2 a write and a read always meet in at least one node:
With W = 1 and R = 1 the read from C asks only C, and C does not have the new write yet. With W = 2 and R = 2 C asks A too and takes v3.
Now the break: C is cut off from A and B.
A and B are a majority and carry on. C cannot reach two nodes and refuses both the read and the write: no answer is better than a wrong one. With W = 1 and R = 1 C answers, but it says 18, while the truth is 17.
This is CAP, stated precisely. While the network is split (P, partition), the node chooses. Either it refuses, so that it does not return an old value or take a write the others do not know about (consistency, C). Or it answers with what it has (availability, A). It cannot do both at once. CAP does not mean “pick two of the three”: the partition is not chosen, it happens. Without a partition the choice is a different one, and PACELC says so: else (E) the database chooses between latency (L) and consistency (C), because synchronous replication and quorums wait on every write.
More precisely, if you are curious
The idea of quorums comes from Amazon's Dynamo paper (2007). Cassandra follows it: every query says how many nodes it needs (ONE, QUORUM, ALL). The versions there are timestamps, and on a conflict the newer write wins (last write wins). The simulator counts versions with a counter. DynamoDB reads are eventually consistent by default, so they can return an old value, and with ConsistentRead they are consistent. Consistency in CAP is not the C in ACID. In CAP it means that every read sees the last confirmed write, as if there were one copy. In ACID it means that the rules of the database hold. CAP was stated by Eric Brewer in 2000 and proven by Gilbert and Lynch in 2002.
The idea
Sharding: each node with its own part
Replication multiplies the reads, but every write still reaches every node, and every node keeps everything. When the data or the writes do not fit on one machine, you split them: each node keeps its own part. The part is called a shard, and the splitting sharding. Which row goes to which shard is decided by the shard key, usually a column such as the customer number.
By range: customers 1 to 7 on A, 8 to 14 on B, 15 and up on C. where says which node a customer is on, and query which shards a query has to ask:
A query over neighbouring numbers asks only two shards. But every new customer has a higher number and goes to C. All of today's sign-ups hit one node, while A and B rest. This is a hot spot.
By hash: a number is computed from the key, and it says the shard. In the simulator the hash is the remainder of dividing by 3:
The new customers go to A, B and C in turn. The price is that a query over a range now asks every shard, because neighbouring numbers are scattered.
And what if the key is the city? The query “the customers from Sofia” will ask one shard. But everyone from Sofia is in one place:
A has 9 of the 20 customers: the six from Sofia, the largest city, plus Burgas and the customer without a city. Their orders and the load go with the customers. The shard cannot be split either: everyone from Sofia has the same key. And a query by number now asks every shard. A good key has many different values and spreads both the data and the load evenly.
One more thing: a table split into shards is no longer in one place. A JOIN of a customer with their orders is fast only if they are on the same shard. That is why orders are usually sharded by the customer number, like the customer. A transaction that writes to two shards needs a distributed transaction (two-phase commit). It is slower, and if its coordinator goes down halfway, the shards wait.
More precisely, if you are curious
In the simulator the three nodes are either copies or shards. In a real database every shard is its own group of copies. In MongoDB every shard is a replica set, and a router (mongos) sends the query to the shard by the key. MongoDB shards by range or by hash, and a piece of the data with a single value of the key that grows too large cannot be split (a jumbo chunk). Cassandra hashes the key with Murmur3 and places the nodes on a ring (consistent hashing): a new node takes over part of the keys without moving all of them. The remainder of dividing by 3 in the simulator is not like that: with a fourth node almost every customer would change shard. The “hash” of a city in the simulator is the city's number in a list. For PostgreSQL sharding comes with the Citus extension, and for MySQL with Vitess.
Exercise
A failure, a quorum and new customers
1. The stock of white cheese is 18 and is on all three nodes. The leader A goes down, and its disk has to be replaced.
A will not be back soon. Make the cluster take writes again, and write the new stock of white cheese: 17.
Solution
elect
write stock:7 = 17 on B
elect, then write stock:7 = 17 on B. B and C have the same data, and on a tie the simulator picks the first by name. Without an election B refuses the write, because it is a follower. start A is no way out: A will not be back.
2. The cluster has no leader and three nodes.
Choose W and R so that a read from any node right after a write sees the new value, and writes still go through when one node is down.
Solution
quorum write 2 read 2
quorum write 2 read 2. 2 + 2 > 3, so a read always meets a node with the last write. When one node is down, two remain, enough for both W and R. With W = 3 writes stop with the first dead node. With W + R ≤ 3 a read can miss the new write, and with R = 3 reads stop.
3. The shop runs an advert and expects hundreds of new customers with numbers 21, 22, 23 and up.
Choose the shard key of the customers so that the new customers are spread over the three nodes instead of going to one.
Solution
shard customers by id hash
shard customers by id hash. By range all the new numbers are above 15 and go to C. By city the new customers from Sofia would still go to one shard, and the simulator does not even know their cities.
The idea
When to use what
- Replication is needed almost always: a copy on another machine protects against a failure, and the followers take the reads. Asynchronous is the default. If losing the last writes is not acceptable (payments), at least one follower is synchronous.
- Reading from a follower is good for the catalogue, the reports and search. It is not good for what the customer has just changed: then the leader reads.
- Sharding is the last step, not the first. A bigger machine, indexes, a cache like Redis and read replicas are enough for most shops. Sharding makes every query and transaction that crosses shards harder.
- CAP is decided for each operation separately. The stock and the payment want consistency. The cart and “recently viewed” can stay available and be a little out of date.
More precisely, if you are curious
Three nodes are the smallest cluster that changes its leader by itself: with two nodes, a node left alone is not a majority. That is why MongoDB recommends an odd number of voting members in a replica set. Google Spanner and CockroachDB run transactions across shards with consensus algorithms (Paxos in Spanner, Raft in CockroachDB). The price is latency on every write, which is PACELC in practice.
Test