d02 — Distributed Versioned Key-Value Store

A fully worked design. The distributed counterpart to the reported coding screen question (../../../research/source-report.md row 7). Doing both is what lets you say "here's my single-node design, and here's exactly which decision breaks when I distribute it" — the sentence that connects the two rounds.

Attempt it yourself first. Nine sections, 45 minutes, then the hostile critique, then the revision.

Table of Contents

• The Prompt
• 1. Requirements and Scope
• 2. Scale Numbers
• 3. API Surface
• 4. Data Model
• 5. High-Level Architecture
• 6. Deep Dive A: Global Version Ordering
• 7. Deep Dive B: Consistent Snapshots Across Shards
• 8. Failure and Recovery
• 9. Bottlenecks and Evolution
• 10. Tradeoffs Explicitly Rejected
• The Hostile Critique
• The Revision
• What Breaks When You Distribute the Coding Answer
• References

The Prompt

"You built a versioned key-value store on one machine. Now make it distributed. Same semantics: every write gets a version, you can read any key as of any past version, and you can take a snapshot that gives you a consistent view. It has to survive machines dying.

Where does it get hard?"

"Where does it get hard?" is the whole question. The answer is not "replication" — that is the easy part. It is that the single-node design rests on two properties that do not survive distribution: a globally ordered version counter, and the ability to take a snapshot by reading one integer.

1. Requirements and Scope

Clarifying questions asked

"Do reads need to be linearizable, or is bounded staleness acceptable?" The single most consequential question here. Assumed: linearizable writes and read-your-writes; snapshot reads may be bounded-stale (a few hundred ms). This selects an architecture; assuming strict serializability everywhere would select a much more expensive one and I would say so.

"Do snapshots need to be consistent across ALL keys, or per-shard?" Assumed globally consistent — that is what makes it a snapshot rather than a scan, and it is where the hard part lives.

"Multi-region?" Assumed single-region multi-AZ, with a note on what changes.

"How long is history retained?" Assumed 7 days or 100 versions per key, whichever is smaller, plus pinning by live snapshots.

Functional

1. put(key, value) -> version, get(key, version=None), delete(key) -> version.
2. snapshot() -> handle; reads through it see a consistent point in time.
3. Multi-key transactions with snapshot isolation.
4. History and compaction.

Non-functional

Explicitly out of scope

• Secondary indexes and range scans by value.
• Cross-region active-active.
• Unbounded history (retention policy above).
• Serializable isolation — snapshot isolation, and I will name the anomaly it permits.

2. Scale Numbers

Traffic. Assume 100k writes/s, 1M reads/s, 10 billion keys, 1 KB values.

Storage. 10¹⁰ × 1 KB = 10 TB logical. With ~5 versions retained per key on average, 50 TB. × 3 replicas = 150 TB. × 1.4 for index and compaction overhead ≈ 210 TB. At 8 TB usable per node that is ~27 nodes for capacity — but see throughput.

Throughput. 100k writes/s across 27 nodes is 3.7k writes/s/node, which is comfortable. 1M reads/s is 37k/node, which is not — that needs either more nodes or a cache. So throughput binds, not capacity: call it 64 nodes, and I would say that the read path is what sizes the cluster.

The number that matters most. 100k writes/s all needing a globally ordered version. If that ordering goes through a single Raft group, every write is a majority round trip: ~1 ms same-DC, so a single group tops out around a few tens of thousands of ops/s. 100k/s does not fit through one sequencer, and that observation is what drives deep dive A.

3. API Surface

PUT    /kv/{key}                 {value} [, if_version]     -> {version}
GET    /kv/{key}                 [?version=V | ?snapshot=S] -> {value, version}
DELETE /kv/{key}                                            -> {version}
POST   /snapshots                                           -> {snapshot_id, version}
DELETE /snapshots/{id}
POST   /txn   {reads: [...], writes: {...}, snapshot?}       -> {version} | 409 Conflict
GET    /kv/{key}/history         [?limit]                   -> [{version, value|DELETED}]

Three choices worth defending:

• if_version is optional compare-and-swap, so a caller can do a single-key CAS without a transaction. It is the cheap 90% case and it costs one column.
• Snapshots are a resource with a lifecycle, not an implicit read mode. They pin history, so they must be releasable and expirable — an abandoned snapshot blocking all compaction is the Postgres long-transaction bloat failure, and I want it to be visible.
• /txn is a single round trip carrying the read set and write set, rather than an interactive session. Interactive transactions hold locks across network round trips, which at 100k/s is a disaster. This is a real limitation and I would state it.

4. Data Model

Shard assignment:  shard = consistent_hash(key)          256 virtual shards
Each shard:        a Raft group of 3 replicas

Per-shard storage (LSM):
  key: (user_key, version DESC)   ->   value | TOMBSTONE

  Reading "as of V" is a seek to (user_key, V) and take the first row —
  a predecessor query, exactly as on one node.

Cluster metadata (a separate, small Raft group):
  shard -> replica set, leader, epoch
  snapshot_id -> pinned_version, created_at, expires_at
  the timestamp oracle's state

Why key order is (user_key, version DESC): the dominant read is "latest, or as of V", and descending version means that is a seek plus one row rather than a scan. In an LSM this also puts all versions of a key adjacent, so compaction can drop superseded versions locally without cross-shard coordination.

Why 256 virtual shards over 64 nodes: more shards than nodes so rebalancing moves a shard at a time rather than splitting; and a power of two so the mapping is cheap. This is consistent hashing with virtual nodes — a node failure spreads its shards across many successors instead of dumping the whole range on one and cascading.

5. High-Level Architecture

                  ┌──────────────┐
   client ───────▶│ Coordinator  │  stateless, autoscaled
                  │  (any node)  │
                  └───┬──────┬───┘
        1. get version│      │ 3. route by consistent_hash(key)
                      ▼      │
        ┌───────────────────┐│
        │ Timestamp oracle  ││  a small Raft group, or HLC per node.
        │  monotonic, global││  DEEP DIVE A
        └───────────────────┘│
                             ▼
   ┌──────────────────────────────────────────────────┐
   │  Shard 0        Shard 1        ...    Shard 255   │
   │  ┌──────────┐   ┌──────────┐          ┌────────┐  │
   │  │Raft: L,F,F│  │Raft: L,F,F│         │  ...   │  │
   │  │ LSM       │  │ LSM       │         │        │  │
   │  └──────────┘   └──────────┘          └────────┘  │
   └──────────────────────────────────────────────────┘
                             ▲
                             │ membership, leases, epochs
                  ┌──────────┴──────────┐
                  │  Metadata Raft group │  small, rarely changing
                  └─────────────────────┘

Consensus per shard, not per cluster. Each shard is its own Raft group, so writes to different shards are independent and the cluster scales horizontally. A single cluster-wide Raft group would cap total write throughput at one leader — the exact cost named in warmup §6.5.

The metadata group holds metadata only — kilobytes, changing rarely. Never data.

6. Deep Dive A: Global Version Ordering

The problem. The single-node design's entire elegance came from a global monotonic counter: a snapshot is one integer, transactions compare one number, and cross-key ordering is free. Distributed, that counter is a consensus problem — and at 100k writes/s it will not fit through one sequencer.

Option 1 — A single Raft-replicated counter (rejected)

Every write calls next_version() on one Raft group.

• ✅ Exactly the single-node semantics. Perfect global order.
• ❌ Every write is a majority round trip before it even reaches its shard. ~1 ms same-DC, so the counter alone caps you well below 100k/s, and it doubles write latency.
• ❌ A single point of failure for every write in the cluster.

Rejected on the arithmetic. If the target were 5k writes/s I would take it, because the semantics are free and the code is trivial. That is the flip condition.

Option 2 — Batched timestamp oracle (viable)

One Raft group, but it hands out ranges. A coordinator asks for 10,000 versions, gets [N, N+10000), and allocates locally.

• ✅ Amortizes the consensus round trip by the batch factor — 100k/s becomes 10 requests/s.
• ✅ Still a total order.
• ❌ Versions are no longer dense or time-ordered. Coordinator A holding [1000, 2000) and B holding [2000, 3000) means a write at real-time T on B can get a higher version than a later write on A. So the order is total but not consistent with real time, which breaks "read as of 10:00" and makes snapshots meaningless as points in time.
• ❌ Gaps when a coordinator dies holding an unused range.

This is the trap, and it is worth walking into deliberately in the interview before backing out of it: batching gives you throughput and silently costs you the property that made versions useful.

Option 3 — Hybrid logical clocks (chosen)

Each node maintains an HLC: a physical component tracking wall clock plus a logical counter for ties.

on local/send event:
    l' = max(l, physical_now)
    if l' == l: c += 1  else: c = 0; l = l'
on receive (l_m, c_m):
    l' = max(l, l_m, physical_now)
    ... take the max, bump the counter appropriately
version = (l, c, node_id)      # node_id breaks remaining ties

• ✅ No coordination on the write path at all. Version assignment is local.
• ✅ Versions stay close to physical time (bounded by clock skew), so "as of 10:00" is meaningful and a snapshot is a real point in time.
• ✅ Respects causality: if A happened-before B then HLC(A) < HLC(B).
• ✅ Constant size — two integers plus a node id — unlike a vector clock.
• ❌ Cannot detect concurrency. Two truly concurrent writes to the same key get an arbitrary but consistent order, which means last-write-wins semantics at the key level.
• ❌ Correctness of snapshots now depends on a bounded clock skew — see deep dive B.

Chosen because the write path is the hot path, and HLCs remove coordination from it entirely while keeping timestamps meaningful. This is what CockroachDB and MongoDB do, and for this reason.

The cost I am accepting, stated plainly: two clients writing the same key at the same instant from different coordinators get an order decided by clock skew rather than by arrival. For a key-value store that is acceptable — it is last-write-wins on genuinely concurrent writes. If a caller needs more, they use if_version (a CAS) or a transaction, both of which go through the shard's Raft leader and are therefore properly ordered.

Ordering within a shard

Within one shard, order comes from Raft, not from the clock: the leader appends writes to its log and the log index is the order. HLC timestamps are what make writes comparable across shards. Both mechanisms are needed and they answer different questions — being able to say that cleanly is the point of this deep dive.

7. Deep Dive B: Consistent Snapshots Across Shards

The problem. On one node, snapshot() was return self._version — an integer, O(1), and trivially consistent. Across 256 shards it is a consistent cut problem: you need a version V such that every shard can answer "what did you look like at V?" and the answers are mutually consistent.

Naively taking hlc_now() on the coordinator fails, and the failure is subtle.

Why the naive version is wrong

t=0    Coordinator C1 takes snapshot S at HLC timestamp 1000.
t=0+ε  Coordinator C2, whose clock runs 3 ms fast, writes key K
       and assigns it HLC timestamp 998 — LOWER than S, because
       C2's HLC was already ahead and this write's physical
       component landed below C1's reading.
       Wait — HLC's max() rule prevents going backwards on a node,
       but C2 never talked to C1, so nothing forced C2 forward.
t=1ms  A read through S hits K's shard and sees version 998 <= 1000.
       So S includes a write that happened AFTER the snapshot was taken.

The snapshot is not a point in time. Reading it twice can even return different answers as in-flight writes with timestamps below S land. Non-repeatable reads through a snapshot — which is exactly the guarantee a snapshot exists to provide.

The fix: pick the snapshot version in the future, then wait

This is Spanner's commit-wait, inverted.

1. snapshot_version = hlc_now() + max_clock_skew        (e.g. now + 250 ms)
2. Register it in the metadata group, so every coordinator learns it and
   advances its own HLC past it (HLC's max() rule then guarantees every
   subsequent write on any node gets a HIGHER timestamp).
3. WAIT until hlc_now() > snapshot_version on the coordinator.
4. Return the handle.

After step 3, no write anywhere can still be assigned a timestamp below snapshot_version — because every node's HLC has been forced past it, and HLC is monotonic per node. The cut is consistent.

The cost: snapshot creation takes max_clock_skew (~250 ms with a conservative NTP bound), which blows my stated 50 ms target. That is a real conflict and I will resolve it in the revision.

What this does NOT cost: it does not block writes. Writes continue throughout; they simply get timestamps above the snapshot version, which is exactly what "after the snapshot" means. That property is worth stating, because interviewers expect a snapshot to be a stop-the-world operation and it is not.

Reading through a snapshot

A read at snapshot_version on a shard must be sure that shard has applied everything up to that version. Two cases:

• The shard leader's HLC is already past it → answer immediately from the LSM with a predecessor seek.
• The shard is behind (a follower, or a leader that has been idle) → it must wait until its HLC advances past the version, or bump its own clock. This is a bounded wait, and it is why an idle shard needs a periodic no-op heartbeat through Raft: otherwise an idle shard's HLC never advances and snapshot reads to it stall.

That heartbeat is the non-obvious operational requirement in this design, and it is the kind of detail that reads as having actually built something.

Compaction under snapshots

Same reachability argument as the single-node version: the pin set is {current_version} ∪ {every live snapshot version}, and an entry survives if it is the predecessor of some pin. The difference is that the pin set is now cluster metadata, so each shard must learn it. It is small and changes rarely, so it rides on the metadata Raft group and shards cache it.

The failure this creates: an abandoned snapshot pins history cluster-wide and blocks all compaction. Hence the expires_at in the metadata, and an alarm on the oldest live snapshot age.

8. Failure and Recovery

Deliberately accepted: two genuinely concurrent writes to the same key from different coordinators are ordered by HLC, i.e. effectively by clock skew, not by arrival. I accept that because the alternative — routing every write through a global sequencer — costs a majority round trip on every write and caps throughput an order of magnitude below the target. Callers who need more use if_version or a transaction, both of which are properly ordered by the shard's Raft log.

9. Bottlenecks and Evolution

What breaks first, in order:

1. The read path, at ~2× current load. 37k reads/s/node against an LSM means a lot of Bloom filter checks and level descents. The fix is a coordinator-side cache for latest-version reads, which is safe because reads are already only read-your-writes rather than linearizable. That turns the read path into a cache hit-rate problem — familiar territory.

2. Cross-shard transactions, at any meaningful rate. My design does single-round-trip transactions with a read set and a write set, but a transaction touching keys on 5 shards needs two-phase commit across 5 Raft groups. That is 2 × (majority RT) plus the coordination, so ~10 ms and a lot of failure modes. At high rates, 2PC across many shards is where this design stops being good. Mitigation: co-locate related keys in one shard by key prefix, so most transactions are single-shard.

3. The metadata group, at ~1,000 snapshot creations/s. Each snapshot is a metadata Raft write. Fix: batch snapshot creation, or make snapshots node-local with a lazily-registered pin.

4. Compaction I/O. Retaining 5 versions per key means compaction reads and writes 5× the logical data. This is LSM write amplification and it is the thing that quietly consumes your disk bandwidth.

At 100×: the fundamental change is multi-region. HLC's skew bound gets much worse across regions, which either forces a much longer snapshot wait or forces you into Spanner's territory — TrueTime hardware and commit-wait on every transaction. I would say plainly that multi-region strong snapshots are a different system, not a scaling of this one.

10. Tradeoffs Explicitly Rejected

Rejected: one global Raft group for everything. Trivially correct, and it caps total cluster write throughput at a single leader — well below 100k/s — while doubling write latency. Flip condition: under ~5k writes/s I would take it for the simplicity.

Rejected: a batched timestamp oracle. Amortizes the consensus cost 10,000×, and silently destroys the property that made versions useful: they stop being consistent with real time, so "as of 10:00" and snapshots become meaningless. Flip condition: if versions only ever needed to be comparable and never temporal, this is strictly better than HLCs and much simpler.

Rejected: vector clocks. They would let me detect concurrent writes rather than silently ordering them, which is a genuine correctness improvement. Rejected because they are O(nodes) in size, must be stored with every value and sent with every message, and pruning entries for departed nodes is subtle enough to be a source of real bugs. Flip condition: if the product needed to surface conflicts to the user (Dynamo-style siblings) rather than resolve them, vector clocks are the right answer and I would pay the cost.

Rejected: last-write-wins by wall clock. Simplest of all. Rejected because a single node with a skewed clock silently wins every conflict forever and deletes other nodes' writes — a documented data-loss mode in Cassandra deployments with clock problems. HLCs give me the same simplicity with a bounded, monitorable relationship to real time.

Rejected: serializable isolation. I provide snapshot isolation, which permits write skew — two transactions read overlapping data, write disjoint keys, both commit, and jointly break an invariant. Rejected because SSI requires tracking read-write dependencies across shards, which is a distributed conflict-detection problem substantially harder than the rest of this design. Flip condition: if callers had cross-key invariants they could not express as a CAS, I would need it — and I would be honest that it changes the system.

Rejected: interactive transactions. Rejected because holding locks across client network round trips at 100k/s is a disaster — one slow client stalls a shard. Single-round-trip transactions with a declared read set are less expressive and vastly more operable.

The Hostile Critique

C1. "Your snapshot takes max_clock_skew — 250 ms by your own number — and your requirements say snapshot creation must be under 50 ms and must not block writes. You've written a design that violates its own stated SLO by 5×. Which number is wrong?"

C2. "You said an idle shard's HLC never advances, so you added a heartbeat through Raft. Every shard, forever, even at 3am with zero traffic. That's 256 Raft groups × 3 replicas doing periodic log appends. What's the steady-state cost of your idle cluster, and what does that do to your LSM?"

C3. "Your snapshot correctness depends on max_clock_skew being an actual bound. NTP gives you a statistical claim, not a bound, and you said so yourself in the failure table. So what actually happens when a node's clock is skewed by more than your bound but not enough to trip your ejection threshold? Walk me through the read."

C4. "You reject vector clocks because they can't detect concurrency, then accept last-write-wins on concurrent writes. But your if_version CAS goes through the shard leader's Raft log. So why isn't every write a CAS? What do I actually lose by making the version check mandatory?"

C5. "Compaction needs the pin set, which lives in the metadata group. The metadata group goes down for an hour. What happens to compaction across 256 shards, and what does the disk usage graph look like?"

C6. "A transaction spans 5 shards. You said 2PC. Walk me through what happens when the coordinator dies between prepare and commit. Who resolves it, and how long are those 5 shards holding locks?"

The Revision

R1 — The snapshot SLO was wrong, not the design (answers C1)

The critique is correct that they conflict, and the honest resolution is that my requirement was wrong, not the mechanism.

Change: split snapshot creation into two operations with different guarantees.

Why this is better than picking one: most snapshot uses genuinely do not need an exact cut, and paying 250 ms for all of them to serve the minority is the wrong default. Exposing the choice — and its cost — is the honest design.

Cost: two modes to document and test, and a user who picks fast and assumes exact semantics gets a subtle bug. Mitigated by making the mode required, with no default, so the caller has to think.

R2 — Heartbeat only what's read (answers C2)

The critique is right and I had not costed it. 256 groups × 3 replicas × a heartbeat every 100 ms is 7,680 log appends/s at idle, all of which enter the LSM, all of which must later be compacted. An idle cluster generating compaction load is a bad design.

Change: do not heartbeat proactively. Instead, advance a shard's HLC on demand:

• A snapshot read arrives at a shard whose HLC is behind the snapshot version.
• The shard leader appends one no-op through Raft to prove it is still leader and to advance its HLC past the requested version, then serves the read.
• The no-op is O(1) per stale shard per snapshot, not per shard per interval.

Cost: the first snapshot read to an idle shard pays one Raft round trip (~1 ms) instead of being free. That is a far better trade than continuous background load, and the cost lands on the operation that actually needs it.

Additionally: cap the no-op rate per shard so a pathological read pattern cannot turn this into the same problem by another route.

R3 — Be honest about the skew bound (answers C3)

The critique exposes an unstated assumption. NTP does not give a bound, so "wait out max_clock_skew" is a probabilistic guarantee dressed as a deterministic one.

Change: make the failure explicit and detectable rather than pretending it cannot happen.

1. Every write carries the writing node's HLC. A shard that receives a write whose timestamp is below a snapshot it has already served reads for rejects it and returns an error to the coordinator, which retries with a fresh timestamp. This converts a silent consistency violation into a retry.
2. Track the worst observed skew as a metric — measured from the HLC max() adjustments each node makes on receive, which is a direct observation of how far ahead other nodes are. Alarm when it approaches the configured bound, and eject well before it exceeds it.
3. Document the guarantee honestly: exact snapshots are consistent provided clock skew stays within the configured bound, the system detects and rejects violations, and the bound is monitored. That is what Spanner buys with atomic clocks and what everyone else approximates.

Cost: a small rejection rate under clock trouble, and an honest guarantee rather than an absolute one. That is the correct trade and it is what CockroachDB does.

R4 — Why not always CAS (answers C4)

Good question, and the answer sharpens the design.

Making every write a CAS would require the client to know the current version, which means a read before every write — turning 100k writes/s into 100k reads + 100k writes, and adding a round trip to the write path. That is the actual cost.

Change: nothing structural, but state the rule explicitly in the API docs and enforce it in the client library:

• Blind write (PUT without if_version) — last-write-wins, ordered by HLC. Use when the value is self-contained and a lost update is acceptable (a cache entry, a heartbeat).
• CAS (PUT with if_version) — properly ordered by Raft. Use when the new value depends on the old.
• Transaction — when the invariant spans keys.

The insight worth stating: the concurrency semantics are a per-write choice, not a system-wide one, and the API should make the caller pick. Defaulting everything to CAS makes the common case slower to protect the uncommon one.

R5 — Compaction must survive metadata loss (answers C5)

The critique is right, and the consequence is worse than it first looks: with compaction stopped across 256 shards under a 100k writes/s workload, disk usage grows at roughly the raw write rate — ~100 GB/hour of un-compacted data. An hour of metadata downtime is a capacity incident.

Change: shards cache the pin set and can compact against a stale one safely.

• The pin set is monotone in a useful direction: a stale pin set contains more pins than the current one (snapshots that have since been released), so compacting against it is conservative — it retains more than necessary, never less.
• So: shards cache the pin set with a TTL, and on metadata unavailability they keep compacting against the last known set. They may retain some garbage; they never drop something a live snapshot needs.
• New snapshots cannot be created while metadata is down, which is correct — a snapshot that no shard knows about is not a snapshot.

Cost: slightly more retained garbage during a metadata outage, reclaimed on the next successful refresh. That is strictly better than stopping compaction.

This is a nice property to have found, and the reasoning generalizes: when a cached authority is unavailable, check whether staleness is conservative in the direction you need. If it is, degrade to the cache instead of stopping.

R6 — 2PC failure handling (answers C6)

The critique found the gap: I said "2PC" without specifying the recovery path, which is where all of 2PC's difficulty lives.

Change: make the transaction record itself Raft-replicated, so no participant depends on the coordinator surviving.

1. The coordinator writes a transaction record — read set, write set, state=PREPARING — into the Raft group of the first shard in a deterministic ordering of the participants. That shard is the transaction's home.
2. Prepare on all participants; each records prepared(txn_id) in its own Raft log and holds locks.
3. The coordinator flips the home record to COMMITTED (or ABORTED). That single Raft write is the commit point — the transaction's outcome is now durable and discoverable independently of the coordinator.
4. Participants apply and release. If a participant does not hear back, it asks the home shard for the outcome.

If the coordinator dies between prepare and commit: the home record is still PREPARING. Participants hold locks. After a timeout, any participant may drive resolution by reading the home record: if still PREPARING past the deadline, it flips it to ABORTED and everyone rolls back. Because the flip is a Raft write, exactly one outcome wins.

Lock hold time: bounded by the resolution timeout, which I set to 5 seconds — long enough that a GC pause does not abort healthy transactions, short enough that a dead coordinator does not stall a shard for minutes.

Cost: one extra Raft write on the commit path (the home record), so a cross-shard transaction is ~3 majority round trips rather than 2. And the honest statement: cross-shard transactions are 5–10× the cost of single-shard ones, which is why the data model should co-locate related keys and why I would surface a metric for cross-shard transaction rate.

What Breaks When You Distribute the Coding Answer

The bridge sentence between the two rounds. Worth having memorized.

The last row is the good news and it is worth ending on: MVCC's core property survives distribution intact. Everything that broke is about ordering and global views, not about concurrency control. That is why the single-node design was worth getting right first.

References

• ../WARMUP.md — every primitive used here from zero
• d01-job-scheduler.md — the other mandatory design, with its own critique
• ../../coding/WARMUP.md#chapter-1-predecessor-queries-and-versioned-state — the single-node version this distributes
• Corbett et al. Spanner: Google's Globally-Distributed Database. OSDI 2012 — TrueTime, commit-wait, the exact problem in §7
• Kulkarni et al. Logical Physical Clocks and Consistent Snapshots in Globally Distributed Databases. OPODIS 2014 — HLC
• Taft et al. CockroachDB: The Resilient Geo-Distributed SQL Database. SIGMOD 2020 — HLCs plus uncertainty intervals in production
• Peng & Dabek. Large-scale Incremental Processing Using Distributed Transactions and Notifications. OSDI 2010 — Percolator's timestamp oracle, i.e. option 1 done properly
• Ongaro & Ousterhout. In Search of an Understandable Consensus Algorithm. USENIX ATC 2014
• Kleppmann, M. Designing Data-Intensive Applications, 2nd ed. — Ch. 7 (snapshot isolation, write skew), Ch. 8 (unreliable clocks), Ch. 9 (linearizability, 2PC)
• O'Neil et al. The Log-Structured Merge-Tree. Acta Informatica, 1996