Read Consistency with Database Replicas at Shopify

Ideas you will meet

Replication

Leader/follower, multi-leader and leaderless replication.

Consistency Models

Linearizable, causal and eventual consistency explained.

Databases

Schemas, indexes, query plans and choosing a store.

What we already know about this

Our own lessons on these ideas. Read them here before you start, so it makes sense.

• Databases Deep Dive · 12 minReplication topologies and multi-region databasesSingle-leader, multi-leader and leaderless replication, how they handle conflicts and lag, and what changes across regions.Read hereClose

  Why replicate

  Replication keeps copies of data on several machines for three reasons: availability (survive a node failure), read scaling (serve reads from replicas) and latency (put data near users). The hard part is keeping copies consistent while writes continue. (Basics are in the replication and sharding lesson of the foundations course; this lesson compares the topologies.)

  Single-leader (primary and replicas)

  One node, the leader (primary or master), accepts writes and ships its changes to followers (replicas), which serve reads.

  100%

  Figure SDA-DDV-04-1 · Single-leader replication · © forcsguy.com

  • Synchronous replication waits for a follower before acknowledging: safer, slower, and a stalled follower can block writes. Asynchronous is faster but a failover can lose the latest writes. A common compromise: one synchronous follower, others asynchronous.
  • Failover must detect failure, choose the most up-to-date replica, redirect clients and ensure the old leader does not come back believing it still leads (split brain).
  • Replication lag causes stale reads. Techniques: read-your-writes (route a user's reads to the leader shortly after they write), monotonic reads (keep a user on one replica), and checking replica position.

  Multi-leader

  Several nodes accept writes (often one per region) and replicate to each other. Benefits: local write latency and continued operation if a region is cut off. Cost: write conflicts.

  Conflict handling options:

  • Avoid them: route all writes for one record to one leader (for example by user or region).
  • Last write wins by timestamp: simple, but silently drops one of two concurrent updates and depends on clocks.
  • Merge values (union of sets, counters) using mergeable data types (CRDTs; see the collaborative editing lesson).
  • Custom resolution in application code, or ask the user.

  Multi-leader also appears in offline-capable apps (each device is a leader) and collaborative editing.

  Before: single leader

  100%

  After: multi-leader across regions

  100%

  What changed

  • Users in each region write to a nearby database
  • Concurrent edits to the same row need conflict handling
  • More moving parts to run and monitor

  Leaderless (Dynamo-style)

  Any replica accepts writes; clients write to several replicas and read from several, using quorums (W + R > N) to get fresh data (see the key-value store lesson). Background anti-entropy and read repair fix stale copies; versions or vector clocks detect conflicts. It favours availability and is typical of Cassandra and DynamoDB-style stores.

  100%

  Figure SDA-DDV-04-2 · Choosing a replication model · © forcsguy.com

  Multi-region designs

  • Active-passive: one region serves writes; another is a warm standby. Simple; failover loses a little data (the replication lag) and takes minutes.
  • Active-active: several regions serve traffic. Requires either partitioning data by home region (each record has one home, cross-region reads are replicas) or true multi-leader conflict handling.
  • Global databases use consensus across regions (see the consensus lesson) for strong consistency at the cost of cross-region write latency; some offer tunable consistency per query.
  • Latency numbers matter: a cross-ocean round trip is around 100 ms or more, so synchronous cross-region replication adds that to every write.
  • Data residency laws may require keeping certain users' data in a region.

  Partitioning and replication together

  Real systems shard data into partitions and replicate each partition (a leader per partition spreads write load across nodes). Rebalancing, hot partitions and cross-partition transactions are the new problems (see sharding and the consistent-hashing lesson).

  Master/slave naming

  You will see "master/slave", "primary/replica" and "leader/follower" for the same single-leader idea; most current documentation prefers the last two.

  Checklist

  • What happens to writes if the leader or a region is lost, and how long?
  • How much data can we lose (RPO) and how long can we be down (RTO)?
  • How do we handle stale reads from replicas?
  • Who resolves conflicts, and how?
  • Have we practised failover?

  Real-world reading

  Related reading

  • Active-Active for Multi-Regional ResiliencyNetflix Engineering · NetflixCase Studies
  • Designing Data-Intensive ApplicationsMartin Kleppmann · O'ReillyBooks
  • Read Consistency with Database Replicas at ShopifyShopify engineers · ShopifyCase Studies
  • Uber Schemaless: A Scalable MySQL DatastoreUber Engineering · UberCase Studies

  Show 6 moreShow fewer

  • Mitigating MySQL Replication Lag and Reducing Read Load at GithubGithub engineers · GithubCase Studies
  • Cassandra architecture overviewApache Cassandra · Apache Software FoundationDocumentation
  • How Discord Stores Trillions of MessagesBo Ingram · DiscordBlogs
  • Spanner: Google's Globally Distributed Database (paper)Corbett et al. · GooglePapers
  • Scaling Multitenant Architecture Across Multiple Data Centres at Shopify - Weingarten, Engineering Lead at ShopifyShopifyYouTube
  • MIT 6.824: Distributed SystemsRobert Morris · MITCourses

  Open this lesson with its quiz and exercise

• Classic System Designs · 14 minDesign a key-value storeA Dynamo-style distributed store: partitioning, replication, quorums, conflict handling and how reads and writes reach disk.Read hereClose

  The problem

  A key-value store supports put(key, value) and get(key). A distributed one must also:

  • Store far more than one machine can hold.
  • Keep serving when machines or networks fail.
  • Have low latency for both reads and writes.
  • Let you tune consistency against availability.

  We use a Dynamo-style design, which is the model behind several widely used stores.

  Partition the data

  Spread keys across nodes with consistent hashing (previous lesson). Virtual nodes keep the load even and let you add capacity gradually. Any node can act as the coordinator for a request, so there is no single leader to bottleneck on.

  Replicate each key

  Copy every key to N nodes: the owner and the next N-1 distinct nodes clockwise on the ring. Spread replicas across racks or zones so one failure does not take out all copies.

  100%

  Figure SDA-CSD-03-1 · Replicating each key to N nodes · © forcsguy.com

  Tunable consistency with quorums

  Choose three numbers:

  • N: number of replicas.
  • W: replicas that must acknowledge a write.
  • R: replicas that must answer a read.

  Setting	Behaviour
  W + R > N	Reads overlap the latest write: strong-ish consistency
  W = 1, R = 1	Fastest and most available, but reads can be stale
  W = N, R = 1	Fast reads, writes fail if any replica is down

  You pick the trade-off per use case, which is what makes this design flexible.

  A quorum write, step by step

  100%

  Figure SDA-CSD-03-2 · Coordinator and a write quorum (N=3, W=2) · © forcsguy.com

  When replicas disagree

  Replicas can hold different versions, for example after a partition.

  • Versioning: attach a version (a vector clock) to each value so the system can tell whether one version supersedes another or they were concurrent.
  • Reconciliation: for concurrent versions, either the application merges them (shopping cart: union the items) or the system uses a simple rule such as last-write-wins, which can silently discard data.

  Two writers at once: conflicts and vector clocks

  100%

  Figure SDA-CSD-03-3 · Concurrent writes create two versions · © forcsguy.com

  100%

  Figure SDA-CSD-03-4 · Vector clocks across writes · © forcsguy.com

  A clock A is an ancestor of B if every counter in A is less than or equal to the matching counter in B. If each has some counter larger than the other's, the versions are siblings and need merging.

  Failure handling

  • Detection: nodes exchange gossip messages about who is alive, so there is no central monitor to fail.
  • Temporary failure: if a replica is down, another node accepts the write and stores a hint, handing it over when the replica returns (hinted handoff).
  • Permanent repair: periodically compare replicas using Merkle trees (hashes of key ranges) and copy only what differs (anti-entropy).
  • Data centre failure: replicate across regions for the keys that need it.

  Noticing failures: gossip

  100%

  Figure SDA-CSD-03-5 · Gossip failure detection · © forcsguy.com

  Staying available through a short outage

  100%

  Figure SDA-CSD-03-6 · Sloppy quorum and hinted handoff · © forcsguy.com

  Repairing replicas for good: Merkle trees

  100%

  Figure SDA-CSD-03-7 · Comparing replicas with Merkle trees · © forcsguy.com

  More failure-handling detail

  • Do not trust one witness. One node saying "that server is down" is not enough; require agreement from several independent sources (gossip gives this naturally: a node is marked down only after others also see its heartbeat counter stall).
  • Sloppy quorum. A strict quorum can block when the usual replicas are down. Instead, take the first healthy servers on the ring for the write or read, and let a stand-in hold the data with a "hint". When the real replica returns, the stand-in hands the data back (hinted handoff). You gain availability and give up the strict overlap guarantee for a while.
  • Replica placement. When walking the ring for N replicas, skip virtual nodes that belong to a server already chosen, and spread replicas across data centers, because nodes in one site often fail together.
  • Bigger machines. Give a more powerful server more virtual nodes so it takes a proportionally larger share.
  • Conflicting versions. Vector clocks grow as more servers touch a key; cap their length and accept that, rarely, old entries dropped make some conflicts harder to untangle.

  Inside one node: how writes and reads hit disk

  100%

  Figure SDA-CSD-03-8 · Write and read path on one node · © forcsguy.com

  Write path

  1. Append to a commit log (sequential, fast, crash-safe).
  2. Apply to an in-memory sorted structure (the memtable).
  3. When the memtable is full, flush it to an immutable sorted file on disk; background compaction merges files.

  Read path

  1. Check the memtable.
  2. If missing, check the files, using a Bloom filter per file to skip files that definitely do not contain the key.
  3. Merge results, newest wins.

  This log-structured design turns random writes into sequential ones, which is why such stores handle huge write volumes.

  One node, many jobs

  100%

  Figure SDA-CSD-03-9 · What every node does · © forcsguy.com

  Summary of the design

  Concern	Technique
  Scale out	Consistent hashing with virtual nodes
  Durability and availability	N replicas across zones
  Consistency dial	Quorums (W, R)
  Conflicts	Vector clocks and merge logic
  Failure detection	Gossip
  Temporary and permanent repair	Hinted handoff, Merkle-tree anti-entropy
  Fast writes	Commit log, memtable, sorted files

  Worked numbers: how many nodes?

  Target: 1 billion keys averaging 1 KB, with 3 replicas, serving 100,000 operations per second.

  • Raw data: 1 B x 1 KB = 1 TB. With three replicas, 3 TB of storage across the cluster, before overhead for indexes and compaction (plan for about 2x headroom on disk for compaction, so around 6 TB).
  • Per-node capacity: if one node comfortably holds 1 TB of data and serves about 10,000 operations per second, 12 to 15 nodes cover both storage (3 TB / 1 TB, with room) and throughput (100,000 / 10,000 x 3 for replication on writes and headroom).
  • With a 20-node ring each node owns about 5% of the keys, and losing one node moves its 5% (about 150 GB of replicas) to its neighbours: repair takes minutes to hours, which is why you keep spare capacity.

  When a replica is down: hinted handoff

  With W = 2 of N = 3, a write must reach two replicas. Suppose node C is down.

  1. The coordinator sends the write to A, B and C; A and B acknowledge, so the write succeeds.
  2. For C, a hint ("this write was meant for C") is stored on a healthy node.
  3. When C returns, the holder of the hint replays the write to C.

  If the node is down for too long the hints are dropped, and a slower anti-entropy repair takes over.

  100%

  Figure SDA-CSD-03-10 · A write while one replica is down · © forcsguy.com

  Anti-entropy with Merkle trees

  Replicas can still drift apart (a hint was lost, a disk was corrupted). Comparing every key between two replicas is far too expensive. A Merkle tree summarises a key range with hashes: each leaf is a hash of a small bucket of keys, and each parent is a hash of its children. Two replicas compare root hashes; if equal, the whole range matches. If not, they compare the children and descend only into the branches that differ, until they find the few buckets that actually disagree and exchange just those keys.

  100%

  Figure SDA-CSD-03-11 · Finding differences without comparing every key · © forcsguy.com

  Mistakes people make

  • Hot keys. One extremely popular key overloads its replicas whatever the cluster size. Cache it in front or split it into several keys.
  • Large values. Multi-megabyte values hurt latency and replication; store big blobs in object storage and keep a pointer.
  • Expecting joins and secondary queries. A key-value store answers "get by key". If you need other queries, add an index structure or use another database.
  • Resolving conflicts with wall-clock time alone. Skewed clocks lose writes; prefer version vectors or let the application merge.
  • Forgetting tombstones. Deletes leave markers that must live long enough to reach all replicas, and be compacted away later. Skipping this resurrects deleted data.
  • Never running repair. Without regular anti-entropy, rarely read data can silently diverge.

  Try it yourself

  Build the consistent hash ring lab to see how few keys move when a node joins, the Bitcask lab for one node's storage engine, and the write-ahead log lab for crash recovery. Then read the consistency and CAP lesson for the quorum rule used above.

  More on this site

  Related reading

  • Amazon Dynamo: Highly Available Key-Value StoreWerner Vogels · AmazonCase Studies
  • The Google File System (paper)Ghemawat, Gobioff and Leung · GooglePapers
  • The Log-Structured Merge-Tree (paper)O'Neil et al. · UMass BostonPapers
  • Time, Clocks, and the Ordering of Events in a Distributed System (paper)Leslie Lamport · Microsoft ResearchPapers

  Show 6 moreShow fewer

  • ZippyDB: Facebook's key-value storeMeta Engineering · MetaCase Studies
  • Inside the Magic PocketDropboxCase Studies
  • Spanner: Google's Globally Distributed Database (paper)Corbett et al. · GooglePapers
  • Read Consistency with Database Replicas at ShopifyShopify engineers · ShopifyCase Studies
  • Designing Large-Scale SystemsSomethingsimilar engineers · SomethingsimilarBlogs
  • Reddit Inc Blog HomepageRedditblog engineers · RedditblogBlogs

  Open this lesson with its quiz and exercise

• Advanced Distributed Systems · 12 minRedis: persistence, replication, Sentinel and ClusterWhat keeps an in-memory database from losing data and from being a single point of failure: snapshots, the append-only file, replicas, Sentinel and hash slots.Read hereClose

  Fast because it is in memory, durable because of logs

  Redis keeps its dataset in RAM (the reason it is so fast; see the Redis and Memcached lesson). To survive a restart it offers two persistence mechanisms, which can be combined.

  Snapshots (RDB)

  Redis periodically writes a point-in-time snapshot of the whole dataset to disk, using a forked child process so the main process keeps serving. Snapshots are compact and quick to load, but anything written since the last snapshot is lost on a crash.

  Append-only file (AOF)

  Redis appends every write command to a log. On restart it replays the log. The fsync policy sets the loss window:

  Policy	Behaviour	Risk
  always	Flush after every command	Slowest, almost no loss
  everysec	Flush once per second	At most about a second lost (common default)
  no	Let the OS decide	Fastest, larger loss

  The AOF grows, so Redis periodically rewrites it into a compact form. Many deployments use AOF with everysec, sometimes plus snapshots for fast backups.

  100%

  Figure SDA-ADS-04-1 · Redis persistence choices · © forcsguy.com

  Replication and Sentinel

  A Redis primary replicates to one or more replicas asynchronously. Replicas serve reads (possibly slightly stale) and can take over. Sentinel is a small set of monitor processes that:

  1. watch the primary and replicas,
  2. agree by quorum that the primary is down,
  3. promote a replica and tell clients where the new primary is.

  100%

  Figure SDA-ADS-04-2 · Sentinel failover · © forcsguy.com

  Because replication is asynchronous, a failover can lose the last few acknowledged writes. Redis offers settings to reduce this (for example requiring a minimum number of replicas to acknowledge), at the cost of availability.

  Redis Cluster: sharding

  When data or traffic outgrow one node, Redis Cluster shards the keyspace into 16,384 hash slots. A key's slot is a hash of the key modulo that number; each node owns a range of slots, and each primary has replicas. Clients learn the slot map and talk to the right node; a wrong guess gets a redirect.

  • Resharding moves slots (and their keys) between nodes while the cluster runs.
  • Multi-key operations only work when all keys are in the same slot. Hash tags (a {tag} part of the key) force related keys into one slot.
  • Failover is built in: replicas of a failed primary are promoted by the cluster.
  • This is the "fixed number of slots" cousin of consistent hashing (see the consistent-hashing lesson).

  Which setup when

  Need	Choice
  Pure cache, loss acceptable	Single node or replicas, little or no persistence
  Must survive restart	AOF (everysec) and snapshots
  Must survive node loss	Replica plus Sentinel
  Data larger than one node, or very high throughput	Redis Cluster

  Also keep memory in mind: set a max memory and an eviction policy (see the caching lessons) so Redis behaves predictably when full, and watch for large keys and slow commands, which block a single-threaded executor.

  Real-world reading

  Related reading

  • Redis DocumentationRedis · Redis Ltd.Documentation
  • Redis persistenceRedis · Redis Ltd.Documentation
  • Redis UniversityRedis · Redis Ltd.Courses
  • Redis BlogRedis engineers · Redis Ltd.Blogs

  Show 6 moreShow fewer

  • Scale with Redis ClusterRedis · Redis Ltd.Documentation
  • RedisRedis · Redis Ltd.GitHub
  • Failover with Redis Sentinel at VintedVinted engineers · VintedCase Studies
  • Memory Optimization in Redis at WattpadWattpad engineers · WattpadCase Studies
  • Speeding Up Redis with Compression at DoorDashDoorDash engineers · DoorDashCase Studies
  • Availability in Globally Distributed Storage Systems at GoogleGooglePapers

  Open this lesson with its quiz and exercise