The previous three lessons took replication for granted: there were replicas, inv-bcn and inv-vlc, that lagged, diverged or reached agreement, but we never explained how data gets from one to the other or what happens along the way. This lesson opens that box. Replication is the mechanism by which the same piece of data is kept on several nodes, and its design directly determines which consistency model you get (03-01), what is sacrificed when a partition strikes (03-02) and where consensus is needed (03-03).

We will first look at why we replicate (availability, latency and read scaling) and then at the three topologies that exist. Leader-follower replication, the most common, with its synchronous, asynchronous and semi-synchronous variants, replication lag and the anomalies it produces (which are exactly the session guarantees of 03-01 seen from the other side), and failover with its dangers. Multi-leader replication, for several regions or offline clients, whose central problem is resolving conflicts between concurrent writes. And Dynamo-style leaderless replication, with write and read quorums, read repair and hinted handoff. The practical part is twofold: we will set up a real PostgreSQL streaming replica for km0_orders in Kilometre Zero's docker-compose.yml, observe it with pg_stat_replication and watch it arrive late; and a Python simulation with N=3 will show when a quorum returns stale reads. The Cassandra database, which implements leaderless replication in production, and partitioning (how to spread different data across nodes, rather than copy the same data) are left for Module 4.

1. Why replicate: availability, latency and read scaling
2. Leader-follower replication
3. Synchronous, asynchronous and semi-synchronous
4. Replication lag and its anomalies
5. Failover and its dangers
6. Multi-leader replication and conflict resolution
7. Leaderless replication: quorums, read repair and hinted handoff
8. Table of the three topologies
9. Hands-on: PostgreSQL streaming replication for km0_orders
10. Simulation: W/R quorum with N=3
11. Common mistakes and tips
12. Exercises
13. Conclusion

Replicating means keeping copies of the same data on several nodes connected by a network. It is done for three reasons, and it is worth knowing which one you are pursuing because each pushes the design in a different direction:

Notice that none of the three goals is "scaling writes": replication copies the same writes to every node, so write capacity does not grow. For that you need partitioning (04-01).

If the data never changed, replicating would mean copying it once. All the difficulty comes from the fact that it changes: every write has to reach every copy, and between reaching one and reaching another there is an interval in which the copies differ. The three topologies below are three answers to the question "who accepts writes and how are they propagated?".

Also called single-leader, primary-replica or master-slave (a term that is falling out of use). It is what PostgreSQL, MySQL, MongoDB, Kafka (per partition) and most systems use:

• One node is the leader (primary). All writes go to it. The first thing it does is write them to its local storage.
• The others are followers (replicas, standbys). The leader sends them a stream of changes (the replication log), and each follower applies them in the same order, so its state converges to the leader's.
• Reads can go to the leader (always current) or to any follower (possibly lagging).

What travels down the stream can be of three kinds, and the difference matters in practice:

To add a new follower without stopping the leader, you take a consistent snapshot of the leader at a point in the log (in PostgreSQL, pg_basebackup), copy it to the follower, and the follower asks the leader for "everything that has happened since that point" (catch-up). Once it has caught up, it follows live. We will do this in section 9.

The key question is when the leader acknowledges the write to the client relative to the moment the follower receives it:

Fully synchronous replication with many followers is impractical: the slowest follower sets the latency of the whole system, and any follower that is down paralyses it. That is why "synchronous" in practice nearly always means semi-synchronous: one synchronous follower (which guarantees the copy) and the rest asynchronous (which serve reads and will be at hand if the synchronous one goes down). PostgreSQL configures this with synchronous_standby_names, and in section 9 we will change it on the fly to see the effect. In the terms of 03-02, asynchronous replication is PC/EL (fast, with lag on the replicas) and synchronous replication is PC/EC (every commit pays the coordination latency).

With asynchronous followers, between the leader acknowledging and the follower applying there is a replication lag: milliseconds under normal conditions, seconds or minutes if the follower is overloaded, if a long query is blocking WAL replay, or after a partition. If reads go to the followers for scaling, that lag turns into the anomalies Anna suffered in 03-01, now with their mechanical cause:

The practical consequence is that "scaling reads with replicas" is not transparent: you have to decide for each query whether it tolerates lag, and measure the lag continuously (pg_stat_replication in section 9; alerts in 07-01). A reasonable limit is to route to replicas only those reads where a lag of a few seconds is harmless (catalogue, order history from days ago, reports) and keep on the leader the reads that precede a decision (is there stock left? has the order been paid?).

If the leader dies, a follower must become the leader: this is failover. Its three steps (detecting that the leader has died, choosing the new leader, redirecting clients) look simple and each one hides a problem:

• Detecting: there is no safe way to tell a dead leader from a slow or isolated one (01-02, 03-02). A short timeout causes unnecessary failovers; a long one prolongs the unavailability.
• Choosing: with asynchronous replication, the most up-to-date follower may not have the last committed transactions: they are lost. Worse: if other systems have already acted on those transactions (an order.created already published by the outbox relay), they are left inconsistent with the database. With semi-synchronous replication, the synchronous follower is the candidate and loses nothing.
• Redirecting: if the old leader comes back believing it is still the leader, there are two nodes accepting writes: split-brain. This is the situation that Raft's terms prevented in 03-03, and in a database without built-in consensus it has to be prevented from outside: by forcibly shutting down the old leader (STONITH, shoot the other node in the head) or with a fencing token.

Precisely because automatic database failover is so delicate, many teams delegate it to tools that rely on a consensus system (Patroni uses etcd for PostgreSQL; MongoDB and Kafka have their own built-in election) and others do it by hand. How a failover is operated, with which checkpoints and how it is rehearsed is the subject of 07-03; here it is enough to know that the choice between asynchronous and semi-synchronous determines how much is lost in it.

In leader-follower, every write goes through a single node. If Kilometre Zero has users in Barcelona and in Valencia and the leader is in Barcelona, every write from Valencia travels 350 km; and if the network between the cities goes down, Valencia cannot write. Multi-leader replication (also master-master) puts a leader in each data centre, which accepts writes locally and replicates them asynchronously to the other leaders. Its legitimate use cases are few and specific:

• Several regions with local writes and tolerance to disconnection between them.
• Offline clients: the mobile app of the van-3 courier records deliveries with no coverage and synchronises when coverage returns. Each device is, in effect, a leader.
• Collaborative editing (several users editing the same document), which is multi-leader with very small replicas.

And its central problem is unavoidable: two leaders can accept concurrent writes to the same data, and when they replicate a conflict appears. We already saw it in the AP simulation of 03-02: Anna reserves on inv-bcn and Mark on inv-vlc. The strategies, from the simplest to the most elaborate:

One detail that tends to surprise people: the topology in which the leaders send each other changes (all-to-all, ring, star) affects the order in which they arrive, and with physical clocks as the LWW timestamp a write that was causally earlier can "win". Serious multi-leader systems use vector clocks or version vectors (01-05) to detect which writes are truly concurrent and which simply arrived late. Multi-leader is powerful and dangerous in equal measure; PostgreSQL does not offer it out of the box (there are extensions such as BDR/pglogical), and the general recommendation is not to use it unless the three use cases above demand it.

The third topology does away with the leader altogether: any replica accepts writes and reads, and coordination is replaced by arithmetic. It was popularised by the Dynamo paper (Amazon, 2007), and it is implemented by Cassandra, Riak, Voldemort and ScyllaDB.

With N replicas of each piece of data, the client (or a coordinator node acting on its behalf) sends each write to all N replicas in parallel and considers it successful once W have acknowledged; and it sends each read to the N replicas (or to a subset) and considers it successful with R responses, keeping the value with the highest version. If

W + R > N

then the set of replicas that acknowledged the write and the set that answered the read overlap in at least one replica, which will have the new version; since the read picks the highest version, it always returns the last acknowledged write. With N=3, the typical configurations are:

The condition W + R > N guarantees that the read sees the last acknowledged write, but it does not give linearizability: two concurrent writes with the same W can end up applied on different replicas, a read during a write in progress may see the new version and the next one the old version, and versions depend on timestamps (LWW, with its problems) or on version vectors. It is "tunable" consistency and in practice fairly strong, but a Dynamo-style system is no substitute for a consensus-backed store for the decisions of 03-03.

Without a leader there is no log to bring a replica that has been down back up to date; instead there are two mechanisms:

• Read repair: when a read receives different versions from several replicas, the coordinator returns the newest and writes it to the replicas that had the old one. Data that is read often repairs itself; data that is not read does not (for that there are background anti-entropy processes that compare replicas using Merkle trees).
• Hinted handoff: if a replica does not respond during a write, another node accepts the write "on its behalf" with a note (a hint) and hands it over when the replica comes back. With this, the write reaches W acknowledgements even with replicas down: this is the sloppy quorum, which increases availability at the price that W + R > N no longer guarantees overlap (the W acknowledgements may come from nodes that are not the N "real" replicas, and a later read from the N real replicas may not see the write until the hint is delivered).

We are going to give km0_orders a real follower. PostgreSQL implements leader-follower by sending its WAL (physical log) over a streaming replication connection; the follower, in hot standby mode, replays the WAL and accepts read-only queries. We add two services to Kilometre Zero's docker-compose.yml (the primary replaces the postgres service of 01-06):

And the script that creates the replication role and authorises the connection, in km0/sql/replication/01-replicator.sh:

What each piece does, line by line:

• wal_level=replica: the primary writes enough information to the WAL for a follower to replay it (the minimal level is not enough; logical would also include what logical replication needs).
• max_wal_senders=5: maximum number of walsender processes, one per follower (or per pg_basebackup in progress).
• wal_keep_size=256MB: how much WAL the primary retains even though it has already applied it itself, in case a follower falls behind. The replication slot (-C -S slot_replica_vlc) is the modern, safer version of this: the primary does not delete WAL that the slot has not yet consumed (with the risk, if the follower disappears for good, of filling the primary's disk).
• pg_basebackup -R: copies the primary's snapshot to the follower's data directory and writes the standby.signal file (which tells it "start as a follower") and the line primary_conninfo = 'host=orders-db-primary user=replicator application_name=replica_vlc ...' in postgresql.auto.conf, which tells it whom to connect to. -X stream receives the WAL generated during the copy over a second connection, so that the snapshot is consistent without blocking the primary. The application_name is the name by which the primary will identify this follower, and we will use it to make it synchronous.
• hot_standby=on on the follower: it accepts read queries while it replays the WAL. Without it, it would be a "warm" follower useful only for failover.
• host replication replicator all scram-sha-256 in pg_hba.conf: authorises the replicator role to open replication connections from any address. In production this is restricted to the replicas' network and encrypted (06-04).

We start up with docker compose up -d orders-db-primary orders-db-replica and, from the primary, query the view that describes each connected follower:

The columns tell the story of section 3: sent_lsn is how far the primary has sent; write_lsn, how far the follower has received; flush_lsn, how far it has written to disk (what counts for durability); replay_lsn, how far it has replayed (what counts for a query to see it). The *_lag columns are the corresponding times: the measured replication lag. sync_state = async confirms that, for now, the primary waits for nobody. From the follower you can see the same thing from the other side:

With a lag of one millisecond it is hard to observe the anomaly, so we are going to provoke it: the follower can pause WAL replay (it will keep receiving and storing it, but will not apply it), which is exactly what happens when a long query on the replica delays replay or when the replica is saturated.

Output:

The third line is the read-your-writes violation with its cause in plain sight: the primary is at LSN 0/30001F8 and the replica is still at 0/3000148. And the final loop is the "minimum version on the client" technique of 03-01 with PostgreSQL's real version: the client stores the LSN of its commit and refuses to read from a replica that has not reached it. Many data access libraries and proxies (pgpool, some ORMs) implement exactly this.

With a parameter change on the fly, the primary starts waiting for the acknowledgement from replica_vlc on every commit:

synchronous_commit (default on) decides what the primary waits for: remote_write (the follower has received it), on (it has written it to disk: zero loss) or remote_apply (it has replayed it: a later read on the replica already sees it, that is, read-your-writes guaranteed in exchange for maximum latency). For a quorum of several followers, the syntax is ANY 1 (replica_vlc, replica_bcn): the semi-synchronous mode of section 3.

Now run the experiment that reveals the price: stop the replica (docker compose stop orders-db-replica) and try an INSERT on the primary. It hangs: the primary cannot acknowledge without the synchronous replica. It is the "synchronous" row of the table in section 3 in the flesh, and the reason why in production you configure ANY 1 of at least two followers, or accept asynchronous replication with its bounded risk of loss (Ctrl+C on the INSERT, go back to synchronous_standby_names = '' and reload to unblock).

To finish, leaderless replication in Python: three replicas, writes to all of them with W acknowledgements, reads from R with selection of the highest version, and read repair. One replica, node-c, is temporarily down during the write, which is the case in which the choice of W and R matters.

Output:

What it shows:

• With W=1, R=1 and with W=2, R=1 (both W + R ≤ N), Anna's write is acknowledged without node-c, and the first read, which lands only on node-c, returns version 1: Mark sees three cheeses when there are two left. It is a stale read that is legal for that configuration. The second read, which touches node-b and node-c, obtains both versions, returns version 2 and repairs node-c; from then on node-c is up to date and the following reads are correct. With W=1, moreover, the write would have succeeded even if only one replica had responded, with everything that implies for durability.
• With W=2, R=2 (W + R > N), a read that only gets a response from node-c does not reach R and fails instead of lying, both the first and the last (a real coordinator would ask another replica instead of failing). Any read with two responses necessarily includes node-a or node-b, which have version 2: there is never a stale read. That is the overlap guaranteed by quorum arithmetic, at the price that every read waits for two replicas and that, if two replicas went down, neither writes nor reads would be possible.

This is the mechanism that Cassandra exposes through the consistency levels ONE, QUORUM and ALL, and with which in 04-04 we will configure Kilometre Zero's historical orders store.

• Adding read replicas without classifying the queries. The "is there stock left?" query that precedes a reservation must not go to an asynchronous replica. Mark in each service's code which reads go to the leader and which go to the replicas, and why.
• Believing that synchronous replication is "safer" and just switching it on. With a single synchronous follower, its failure blocks every write in the system. If you need zero loss, configure ANY 1 over at least two followers and monitor that they are alive.
• Trusting automatic failover with no protection against split-brain. An old leader that comes back and keeps accepting writes is worse than an outage. Use a consensus-backed tool (Patroni + etcd) or a fencing token, and rehearse the failover (07-03, 07-06).
• Forgotten replication slots. A slot whose follower has disappeared retains WAL for ever and fills the primary's disk. Monitor pg_replication_slots and drop the ones that are not in use.
• Multi-leader with LWW and physical clocks "because that is the default". It is the combination most certain to lose data (03-02). If you need multi-leader, decide the conflict resolution per data type and use version vectors to detect real concurrency.
• A sloppy quorum without knowing it. Cassandra and Riak have the sloppy quorum and hinted handoff enabled by default in some configurations; W + R > N stops guaranteeing overlap. Read the documentation for your version before reasoning with the arithmetic of section 7.
• Tip: measure replication lag as a first-class metric (replay_lag, or the LSN difference converted with pg_wal_lsn_diff) and define a threshold above which the load balancer stops sending reads to that replica.
• Tip: for read-your-writes on PostgreSQL, the commit LSN is the best "version number" there is: it is monotonic, it is already computed and the replicas know exactly where they stand relative to it.

Kilometre Zero has one PostgreSQL database per service. For km0_orders, km0_inventory, km0_payments and the catalogue database, decide between asynchronous, semi-synchronous (ANY 1 of two followers) and remote_apply, state which reads you would send to the replicas and what maximum data loss you accept in a failover. Justify each choice with the concepts of this lesson and of 03-02.

Write a function read_order(order_id, min_lsn) for the orders service that receives the LSN of the user's last commit (stored in their session) and returns the order from the replica if it has reached that LSN, or from the primary if not, recording which of the two it used. Use psycopg and the PostgreSQL functions seen in section 9. How does the service obtain and store the LSN after each of the user's writes?

Add a sloppy quorum mode to LeaderlessStore: if a replica is down during the write, an auxiliary node node-h accepts the write with a hint for it, and delivers it when the replica comes back (deliver_hints()). Reproduce the scenario with W=2, R=2 and show that, between node-c coming back and the hint being delivered, a read from node-b and node-c... can it be stale? Reason about what W + R > N does and does not guarantee with a sloppy quorum, and what is gained in exchange.

Solution 1:

• km0_payments: semi-synchronous ANY 1 with synchronous_commit = on. A charge that is recorded and then lost in a failover is money taken with no order (or vice versa); the acceptable loss is zero. Reads to replicas: only reports and accounting reconciliation (harmless lag); the status of a payment always from the primary. It is the "CP" row of 03-02.
• km0_orders: semi-synchronous ANY 1. An order confirmed to Anna that disappears in the failover, with its order.created already published by the outbox relay, produces an orphan event that inventory and payments would process for a non-existent order. Acceptable loss: zero. Reads to replicas: "my orders" with a minimum LSN (exercise 2) and history more than a day old.
• km0_inventory: semi-synchronous ANY 1 for reservations (the last unit, 03-02); stock reads for the catalogue can go to asynchronous replicas (the catalogue is already eventual). Acceptable loss: no confirmed reservation; stock.updated events go through the outbox and follow the same reasoning as orders.
• Catalogue: asynchronous. A price or description change lost in a failover is re-entered by the producer; there is no irreversible event downstream. Acceptable loss: the last few seconds of changes. All reads to replicas, with no LSN (the effective price is fixed in orders).

remote_apply is not worth it in any of them: it doubles write latency to obtain read-your-writes on the replica, which can be had more cheaply with the LSN on the client.

Solution 2:

The LSN is obtained with pg_current_wal_lsn() after the commit, on the same connection, and stored in the user's session (signed cookie, session Redis or the session token). A user who has never written (min_lsn is None) goes to the primary to be on the safe side, or to the replica if the read tolerates lag; one improvement is to use pg_last_wal_replay_lsn() compared with the LSN in a single query and perform the read only if the replica is up to date, as here. Note that the session's LSN must be the maximum of the LSNs of all the user's writes (monotonic reads), and that a failover to a replica whose history has forked invalidates the earlier LSNs, another reason for semi-synchronous replication.

Solution 3:

Outline of the implementation: in write, when a replica returns False, (key, value, version) is stored in self.hints[replica_name] and the auxiliary node's ack is counted; deliver_hints() goes through the hints and calls write on the replicas that are available again. With W=2, R=2: Anna's write is acknowledged by node-a, node-b and the hint for node-c (three acks, although only two are real replicas). Once node-c is back and before deliver_hints(), a read from node-b and node-c obtains v2 from node-b and v1 from node-c: not stale, because node-b is a real replica that has the write. But imagine that node-b had also been down: the acks would be node-a plus two hints, W=2 would be met, and a later read from node-b and node-c (both back, with no hints delivered) would return v1 with R=2 satisfied: a stale read with W + R > N. What W + R > N guarantees with a sloppy quorum is that the write is on W nodes of some kind (durability) and that it will be delivered to the real replicas when they come back (convergence); what it no longer guarantees is the overlap between the replicas that acknowledged and those that are read. In exchange, the write succeeds even with most of the real replicas down: maximum availability, the AP choice in its purest form.

Replicating means keeping copies of the same data on several nodes to gain availability, latency and read scaling (never write scaling), and this lesson has gone through the three ways of doing it. In leader-follower, all writes go through one node that sends them in order to the others; the choice between synchronous, asynchronous and semi-synchronous decides how much is lost if the leader dies and how much each commit costs, and the followers' replication lag is the mechanical cause of the session anomalies of 03-01, which are avoided by routing to the leader or waiting for the LSN. Failover turns a follower into the leader with three dangers (detection, loss of transactions, split-brain) that we leave for 07-03. In multi-leader, every region accepts writes and the problem is conflict, with a repertoire of solutions ranging from avoiding it to LWW, merges, CRDTs and application-level resolution. In leaderless replication, the arithmetic W + R > N replaces coordination, with read repair and hinted handoff to keep the replicas up to date and the sloppy quorum as a concession to availability. The hands-on part was real: km0_orders now has a PostgreSQL streaming follower created with pg_basebackup -R, observed with pg_stat_replication, paused to watch Lucy fail to find her order, and waited for until the right LSN; and the simulation with N=3 showed Mark reading three cheeses when there were two left whenever W + R ≤ N.

With this we know how to copy the data of one service. But the most important operation at Kilometre Zero, creating an order, no longer touches a single database: it inserts into km0_orders, deducts in km0_inventory and charges in km0_payments, three databases belonging to three services, each replicated as we have just seen, and no PostgreSQL transaction spans all three. If the charge fails after the stock has been deducted, who puts it back? If orders crashes between the deduction and the charge, what state is Mark's order left in? What the monolith solved with a BEGIN and a COMMIT now needs an atomic commit protocol or, more commonly in microservices, a saga with compensations. That is the subject of the last lesson of the module: distributed transactions and sagas.