Time, Clocks, and the Ordering of Events in a Distributed System (paper)

Ideas you will meet

Time and Ordering

Logical clocks, vector clocks and bounded clock error.

Distributed Systems

Consensus, time, failure and the fundamentals of scale-out.

What we already know about this

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

• Classic System Designs · 10 minDesign consistent hashingHow to spread keys over servers so that adding or removing a server moves only a small share of them.Read hereClose

  The problem

  You have a cluster of cache or storage servers and need a rule that says which server holds which key. The first idea is server = hash(key) mod N. It works until N changes, which it always does: a server fails, or you add capacity.

  100%

  Figure SDA-CSD-02-1 · Why mod-N hashing breaks when N changes · © forcsguy.com

  When N changes, almost every key maps somewhere new. A cache suddenly misses on nearly everything and the database behind it is flooded. A storage cluster would have to shuffle most of its data.

  Goal: when a server joins or leaves, only about 1/N of the keys should move.

  The idea: a hash ring

  1. Imagine hash values arranged in a circle (the ring), from 0 up to a maximum and wrapping around.
  2. Hash each server onto the ring.
  3. Hash each key onto the same ring.
  4. A key belongs to the first server clockwise from its position.

  100%

  Figure SDA-CSD-02-2 · Hash ring ownership · © forcsguy.com

  Adding a server D between A and B takes over only the keys that fall between A and D. Everything else stays where it is. Removing a server hands its keys to its clockwise neighbour. Either way, only the neighbouring arc moves.

  Before: hash mod N

  100%

  After: consistent hashing ring

  100%

  What changed

  • Adding a server moves only a small slice of keys
  • Cache hit rate does not collapse after a change
  • Virtual nodes even out the load

  Finding the owner efficiently

  Keep the server positions in a sorted list. To place a key, hash it and use a binary search for the first position greater than or equal to it, wrapping to the first server if you pass the end. That is O(log N) per lookup.

  Two problems with the basic ring

  100%

  Figure SDA-CSD-02-3 · Why a plain ring can be unfair · © forcsguy.com

  Fixing uneven load: virtual nodes

  With only a handful of servers, the arcs between them are uneven, so some servers hold far more keys. And when a server dies, all its load lands on a single neighbour.

  The fix is virtual nodes: each physical server is hashed at many points (say 100 to 200) on the ring.

  100%

  Figure SDA-CSD-02-4 · Virtual nodes per server · © forcsguy.com

  • Keys spread much more evenly.
  • When a server leaves, its load is shared among many others.
  • A bigger server can be given more virtual nodes than a smaller one.

  The cost is a larger ring to store and update.

  Which keys move, exactly

  When a server joins, find the previous server going anticlockwise from it; only the keys between that neighbour and the new server move, and they all move to the new server. When a server leaves, the keys it held (the span back to its anticlockwise neighbour) move to the next server clockwise. Nothing else is touched, which is why the data transfer stays small and predictable.

  Spreading keys evenly also reduces hotspots: without it, several very popular keys can land on the same server and overload it. With many virtual nodes per server, the spread gets tighter (published experiments show the imbalance shrinking from around 10 percent to about 5 percent as virtual nodes per server go from roughly 100 to 200), at the price of more ring metadata to store, so the count is a tuning knob.

  100%

  Figure SDA-CSD-02-5 · Adding a server to the ring · © forcsguy.com

  100%

  Figure SDA-CSD-02-6 · Removing a server from the ring · © forcsguy.com

  Where it is used

  • Distributed caches (so adding a node does not wipe the cache).
  • Partitioned databases such as Dynamo-style stores, where each key is also copied to the next few servers clockwise for replication (next lesson).
  • Load balancing with stickiness, where the same client keeps hitting the same backend.
  • Content delivery to decide which edge server holds an object.

  Things to watch

  • Hot keys: one very popular key still loads its owner. Replicate it or add a local cache.
  • Choose a good hash function: uniform distribution matters more than cryptographic strength.
  • Membership changes must be learned by all clients consistently, through a config service or gossip, otherwise two clients disagree about the owner.

  More on this site

  Related reading

  • Consistent Hashing and Random Trees (paper)Karger et al. · MITPapers
  • NGINX HTTP Load BalancingNGINX · F5 / NGINXDocumentation
  • Time, Clocks, and the Ordering of Events in a Distributed System (paper)Leslie Lamport · Microsoft ResearchPapers
  • Clock Skew when Syncing Password Payloads at DrobboxDrobbox engineers · DrobboxCase Studies

  Show 6 moreShow fewer

  • The System Design PrimerDonne Martin · donnemartinGitHub
  • Solving Remote Build Cache Misses (2 parts) at SoundCloudSoundCloud engineers · SoundCloudCase Studies
  • Load Balancing infrastructure to support more than 1.3 billion users at FacebookFacebook engineers · FacebookCase Studies
  • Meta Engineering BlogMeta Engineering · MetaBlogs
  • XING Engineering BlogXING engineers · XINGBlogs
  • Gusto Engineering BlogGusto engineers · GustoBlogs

  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

• Core Concepts: How Things Actually Work · 8 minTime, clocks and ordering of eventsWhy you cannot trust timestamps across machines, and how logical clocks give a safe ordering.Read hereClose

  Why time is hard

  On one machine, order is easy: the clock only goes forward (mostly). Across machines, there is no shared clock.

  • Each machine's clock drifts, and synchronisation (NTP) is only accurate to some milliseconds, sometimes much worse.
  • Clocks can jump (corrections, leap seconds, virtual machine pauses).
  • Messages take unpredictable time, so "A happened before B" cannot be read from two clocks.

  So never use wall-clock timestamps from different machines to order important events (or to decide who wins a conflict, or to measure the timeout of a lock).

  Image: National Institute of Standards and Technology (NIST) · Public domain · Source · Wikimedia Commons: NIST-F2 cesium fountain atomic clock

  What we can know: happened-before

  Event A happened before B if A could have caused B: same process in order, or a message sent at A and received at B (and transitivity). Events with no such chain are concurrent.

  Logical clocks

  Lamport clock

  Each process keeps a counter. Increase it on each event; attach it to messages; on receiving, set it to max(local, received) + 1. Guarantee: if A happened before B, then clock(A) < clock(B). The reverse does not hold, so equal-looking order does not prove causality.

  Vector clock

  Each process keeps a vector of counters, one per process. Comparing two vectors tells you whether one event happened before the other or whether they are concurrent. Used by Dynamo-style stores to detect conflicting writes.

  100%

  Figure SDA-CORE-17-1 · Vector clock · © forcsguy.com

  Practical choices

  Need	Tool
  A total order of events	A single leader, or consensus (Raft, Paxos), or a log such as Kafka partitions
  Detect concurrent updates	Version vectors, or version numbers with conditional writes
  Merge concurrent edits	CRDTs, operational transformation, application-level merge
  Unique, roughly time-ordered ids	Snowflake-style ids (accepting small skew)
  Real-time bounds	Databases that use atomic clocks and wait out uncertainty (such as Spanner)

  Rules of thumb

  • Use monotonic clocks (not wall time) to measure durations on one machine.
  • Treat timestamps as approximate and for humans; use versions for correctness.
  • Make operations idempotent and commutative where possible, so arrival order matters less.

  More on this site

  Related reading

  • Time, Clocks, and the Ordering of Events in a Distributed System (paper)Leslie Lamport · Microsoft ResearchPapers
  • Spanner: Google's Globally Distributed Database (paper)Corbett et al. · GooglePapers
  • The Raft Consensus Algorithm (site and visualisation)Diego Ongaro and community · raft.github.ioDocumentation
  • In Search of an Understandable Consensus Algorithm (Raft paper)Ongaro and Ousterhout · Stanford UniversityPapers

  Show 6 moreShow fewer

  • How Ringpop from Uber Engineering Helps Distribute Your ApplicationUber Engineering · UberCase Studies
  • Google SRE Book: Managing Critical State (Distributed Consensus)Google SRE team · GoogleBlogs
  • The Chubby Lock Service (paper)Mike Burrows · GooglePapers
  • Clock Skew when Syncing Password Payloads at DrobboxDrobbox engineers · DrobboxCase Studies
  • Airflow at GrabGrab engineers · GrabCase Studies
  • OpenTable BlogOpenTable engineers · OpenTableBlogs

  Open this lesson with its quiz and exercise