Distributed Data

Distributed data means one logical dataset is stored on more than one machine, either as several copies of the same rows (replication), as different slices of the rows (partitioning), or both. Once that is true, a read or write is no longer one local operation. It is a set of messages between machines that can be slow, reordered, duplicated or lost, and the database has to say what a client is promised when some of those messages do not arrive. The central engineering question is not what the feature does when everything is healthy. It is what a given operation guarantees under concurrency, failures and scale.

That promise is what consistency in a distributed database means: whether the replicas of a piece of data agree on its value and on the order in which it changed, and what a reader is allowed to see while they do not yet agree. It is a spectrum rather than a switch. At the strong end, every read reflects the latest acknowledged write, at the cost of coordinating replicas and sometimes waiting or failing during a partition. At the eventual end, replicas accept updates independently and converge later, which is faster and more available but lets readers see stale or conflicting values in the meantime.

A useful mental model is two independent dials. Partitioning decides which machines own a key: a router hashes or ranges the key to a shard so no single machine has to hold or serve everything. Replication decides how many copies of that shard exist and how they agree: one leader orders the writes and followers receive them, or any replica may accept writes and they are reconciled later. Most of the difficulty is in the second dial, because every extra copy is a place where the data can be temporarily or permanently different.

The rest of this lesson follows that second dial. It starts with what disagreement looks like to a user, sharpens the vague words "strong" and "eventual" into definitions you can test, shows what CAP does and does not say, and then works through the mechanisms that decide who is right: quorums, leader-based consensus, conflict resolution and clocks. It closes with a production problem, selling the last seat in two regions at once, where none of the mechanisms is enough without a business rule. The mechanics of replicas, pooling and sharding as scaling tools are in Scaling; the single-node transaction semantics that these protocols try to extend are in Transactions & Concurrency.

A customer updates their profile in region A while region B is briefly unreachable. Which copy is authoritative? Can users in B still read? Can they write? What happens to those writes when connectivity returns? A distributed database has to answer all four, and the answers are trade-offs among availability, latency, consistency and tolerance of failure. A label such as "strongly consistent" or "eventually consistent" answers none of them until it names the operation, the scope (one key, one partition, the whole cluster, one region) and the failure model it holds under.

The most common way users meet the problem has nothing to do with partitions. It is ordinary replication lag. With asynchronous replication the primary commits, answers the client, and ships the change to replicas afterwards. For some interval, which is usually milliseconds but can be seconds or minutes during a write burst, a replica is behind. A user who saves a change and is then routed to that replica sees the old value and concludes the save failed, resubmits, or files a ticket.

The three strategies in the figure fix the same symptom with different costs. Pinning a session to the primary for a short window is cheap to build but the window is a guess about lag. A position token is an exact condition: the client carries the log position of its own last write, and a replica serves the read only once it has applied that position. It costs a little latency when the replica is behind and needs a fallback to the primary when it is far behind. Routing every read to the primary removes the problem and also removes the read scaling that replicas were added for. None of these is a general consistency guarantee; they implement one session guarantee, read-your-writes, which the next section defines.

A concrete position token in PostgreSQL

PostgreSQL streaming replication numbers its write-ahead log with log sequence numbers (LSNs), which makes a token easy to build. The primary reports its current LSN after a commit, and the standby reports how far it has replayed. The query below is the same mechanism as the third strategy; the LSN literal is an example, not output.

-- PostgreSQL 10+. Read-your-writes on a streaming standby using a WAL position (LSN) token.

-- 1. On the primary, right after the write commits, capture a position at or past the commit:
SELECT pg_current_wal_lsn();            -- e.g. '0/3000A28'; return it to the client as an opaque token

-- 2. On a standby, before serving that client's read, check whether replay has reached the token:
SELECT pg_wal_lsn_diff(pg_last_wal_replay_lsn(), '0/3000A28') >= 0 AS caught_up;
-- true  -> safe to read here
-- false -> wait a few milliseconds and re-check, or route this read to the primary

-- 3. Monitoring on the primary: how far behind is each standby?
SELECT application_name, state, sync_state,
       write_lag, flush_lag, replay_lag,                                  -- intervals (PostgreSQL 10+)
       pg_wal_lsn_diff(pg_current_wal_lsn(), replay_lsn) AS replay_bytes_behind
  FROM pg_stat_replication;

The same view is the monitoring signal. Alert on replay lag in time and in bytes, because a replica can show small time lag while a long-running query on it delays replay, and a replica can show large byte lag right after a burst that it will clear in seconds. Lag is also what turns an asynchronous failover into data loss, which the section on leaders and failover returns to.

The short version of strong vs eventual consistency: a strongly consistent read returns the most recent acknowledged write, as if there were one copy of the data, and the system pays for that with cross-replica coordination (higher latency, reduced availability under partition). An eventually consistent read may return an older value, and the system promises only that replicas converge once updates stop. Real systems offer many points between the two, which the definitions below separate.

Linearizability is a property of operations on a single object, such as one register or one key. Each operation appears to take effect atomically at some instant between its invocation and its response, and that instant order respects real time: if operation A finished before operation B began, B cannot be ordered before A. Serializability is a property of transactions over many objects: the outcome equals that of some serial order of the transactions, but nothing requires that order to match real time. A serializable database may legally order a transaction that started later, and returned later, before an earlier one. Strict serializability is the combination, and it is the strongest of the three and the most expensive to provide across machines.

Snapshot isolation gives every transaction a consistent snapshot of committed data and rejects write-write conflicts on the same row, but it can still allow write skew, where two transactions read overlapping data and write disjoint rows so that a rule over the combination breaks. In PostgreSQL the REPEATABLE READ level is snapshot isolation, and SERIALIZABLE adds detection of the dangerous structures. That is a single-node isolation story; see Transactions & Concurrency for the anomalies in detail. It matters here because a distributed database that advertises snapshot isolation is promising something about transactions and not about real-time freshness of reads.

Eventual consistency is a convergence property: if updates stop and communication resumes under the system's assumptions, all replicas end up with the same value. It does not bound how long that takes, it does not say what a read returns in the meantime, and it does not preserve business invariants. Two replicas that accept conflicting writes can converge to a value that is identical and wrong.

The figure makes the definition checkable. Take any recorded history of reads and writes on one register and ask whether a single order exists in which each operation takes effect between its start and end and each read returns the latest preceding write. The middle history fails because the read began after the write had been acknowledged. The third fails in a subtler way that matters for the quorum section: while a write is still in flight a read may return either value, but once one reader has returned the new value no later reader may return the old one.

Session guarantees

Many products do not need a global guarantee. They need the four session guarantees, which constrain what one client sees across its own requests:

• Read-your-writes: after a client writes a value, its later reads see that write or a newer one.
• Monotonic reads: once a client has seen a value, its later reads never return an older state of that object.
• Monotonic writes: a client's writes are applied in the order the client issued them.
• Writes-follow-reads: a write issued after reading a value is ordered after the write that produced that value, so a reply never appears before the post it answers.

These are cheap compared with linearizability because they are scoped to one session and can be enforced with sticky routing to one replica or with a version or position token carried by the client. A globally consistent account balance and a possibly stale public like count do not need the same machinery. Choose the weakest guarantee that keeps the user-visible behavior correct, per operation, and write it down in the API documentation so that clients know which reads may be stale.

The CAP result, proved by Gilbert and Lynch, says that in an asynchronous network that can drop messages, a service cannot at once guarantee linearizability and a non-error response from every non-failed node when the network is partitioned. Each word has a formal meaning that differs from everyday usage. "Consistency" is linearizability, not the C in ACID. "Availability" means every request that reaches a working node eventually gets a successful response, including a node stranded on the minority side; it is not the same as uptime, and a service where the majority side keeps working while the minority returns errors is, in CAP's terms, not available. "Partition tolerance" is not a feature you can opt out of: networks do partition, so the real choice is what to give up when one happens.

During the partition each side has to choose. Refuse requests that cannot be answered consistently, as in a majority-quorum design, and the minority side is down for as long as the partition lasts. Accept writes on both sides, and the system stays available but you now owe a merge and may have violated an invariant in the meantime. A middle path serves possibly stale reads from the minority while refusing writes. None of the three is a flaw. They are three different products, and the right one depends on the operation: a payment ledger and a presence indicator should not share a policy.

CAP says nothing about the system when there is no partition, and that is where most of the time is spent. There, the trade is latency against consistency: waiting for more replicas before acknowledging gives stronger guarantees and slower writes, and answering from the nearest copy is fast and possibly stale. Abadi's PACELC formulation names both regimes. Avoid describing a database as permanently choosing two of three letters. Most real systems let you pick per operation, for example a consistency level per query, and the same cluster can behave very differently under those settings.

In a simple replicated register with N replicas, a write waits for W acknowledgments and a read consults R replicas. If R + W > N, every read quorum must share at least one replica with every write quorum, because two sets whose sizes sum to more than the population cannot be disjoint. With N=3, R=2 and W=2 every pair of two replicas out of three intersects, and the cluster tolerates one unreachable replica for both reads and writes. The shared replica holds the newest acknowledged value, and a reader that takes the highest version among its R answers will find it.

The first scenario is the intended case. The write reaches A and B while C is partitioned, the client is told it succeeded, the partition heals, and a read of B and C resolves the disagreement because B is in both quorums. The second scenario is why overlap alone is not a consistency model. A write that reaches fewer than W replicas is reported as failed but is not undone, so the replica that took it keeps it. A later read that includes that replica returns the failed write; a read that avoids it returns the old value. The overlap guarantee is about acknowledged writes. Failed, timed-out and in-flight writes are indeterminate, which is exactly the case in the third history of the consistency section.

The calculator shows the availability arithmetic. Raising W makes writes stronger and read-light but means a write cannot proceed when more than N−W replicas are unreachable; raising R does the opposite. W=N with R=1 gives fast reads and writes that stall on any single failure. W=1 with R=N is the mirror. N=3 with R=W=2 is the common balance. Setting 2W > N additionally means two conflicting writes cannot both reach disjoint quorums, which matters when a protocol relies on it to detect concurrent writers. Latency follows the same rule: a quorum operation completes at the speed of the R-th or W-th fastest replica, so tail latency depends on slow replicas more than on the median.

What the arithmetic leaves out

For an overlap to be useful, the protocol around it has to be right. It needs a way to order versions: if two clients write concurrently, which one does a reader prefer, and by what comparison? A wall-clock timestamp can lose a write (a later section shows how); a per-key version assigned by a single coordinator or a consensus log cannot. It needs to deal with sloppy quorums and hinted handoff, where a write that cannot reach the key's owners is accepted by other nodes to preserve availability and forwarded later. Those substitute nodes are outside the replica set a read consults, so R + W > N no longer implies an intersection until the hints are delivered. It needs read repair or anti-entropy so stale replicas do converge, and ideally write-back before a read returns so that a read cannot be followed by an older read.

// Node.js (ES modules). A versioned register on N=3 replicas. A write succeeds only if W replicas
// acknowledge; a read asks R replicas and returns the answer with the highest version.
class Replica {
  constructor(name) { this.name = name; this.up = true; this.version = 0; this.value = null; }
  store(version, value) {
    if (!this.up) throw new Error(this.name + ' unreachable');
    if (version > this.version) { this.version = version; this.value = value; }
  }
  load() {
    if (!this.up) throw new Error(this.name + ' unreachable');
    return { version: this.version, value: this.value };
  }
}
const attempt = (fn) => { try { return fn(); } catch { return null; } };

function write(replicas, W, version, value) {
  const acks = replicas.filter((r) => attempt(() => (r.store(version, value), true))).length;
  if (acks < W) throw new Error('write v' + version + ': ' + acks + '/' + W + ' acks');
}
function read(replicas, R) {
  const answers = replicas.map((r) => attempt(() => r.load())).filter(Boolean).slice(0, R);
  if (answers.length < R) throw new Error('read: ' + answers.length + '/' + R + ' answers');
  return answers.reduce((best, a) => (a.version > best.version ? a : best));
}

const [A, B, C] = ['A', 'B', 'C'].map((n) => new Replica(n));
const cluster = [A, B, C];
write(cluster, 2, 1, 'v1');

C.up = false;                         // C is cut off
write(cluster, 2, 2, 'v2');           // acknowledged by A and B: W=2 is met
C.up = true; A.up = false;            // partition heals, but now A is the unreachable one
console.log(read(cluster, 2));        // { version: 2, value: 'v2' }  B has it, C is stale
A.up = true;

// A write that reaches fewer than W replicas is reported as failed, but is not rolled back.
B.up = false; C.up = false;
try { write(cluster, 2, 3, 'v3'); } catch (e) { console.log(e.message); }   // write v3: 1/2 acks
B.up = true;
console.log(read(cluster, 2));        // { version: 3, value: 'v3' }  the "failed" write is visible

A.up = false; C.up = true;            // a later read that happens to avoid A
console.log(read(cluster, 2));        // { version: 2, value: 'v2' }  the value went backwards

Dynamo-style leaderless stores and consensus-backed stores therefore have materially different guarantees even when both are described with quorums. A leaderless store exposes tunable R and W, offers eventual or session-level behavior by default, and typically needs a separate conditional-write mechanism, which some systems implement with a consensus protocol, when a compare-and-set must be linearizable. A consensus-backed store orders every write through a replicated log, so a majority-acknowledged write has a single place in the order. Read the documentation of the specific system for what its quorum levels do under concurrent writes and node failure before you treat QUORUM as a synonym for linearizable.

Leader-based replication routes every write for a partition through one elected leader, which assigns it a position in an ordered log and sends the log to followers. Consensus protocols such as Raft and Multi-Paxos turn this into an agreed, replicated log under stated assumptions: a majority of nodes is reachable and functioning, messages may be lost or delayed but are not corrupted, and leadership is tied to a monotonically increasing term (Raft) or ballot (Paxos). An entry is committed once the leader knows a majority has stored it, and any future leader is constrained so it must already contain every committed entry. Under Raft a candidate whose log is behind cannot win a majority of votes, which is what carries committed writes across a leader change.

Three details in the figure carry most of the correctness. A follower acknowledges an entry only after it has made the entry durable, otherwise "a majority has it" would not survive a simultaneous reboot of that majority. The term number travels with every message, so an old leader that wakes up after a partition learns it has been replaced and steps down instead of continuing to act. And the old leader's uncommitted entry is simply overwritten by the new leader's log, which is correct because it was never acknowledged as committed. The client that sent it, however, saw a timeout or an error, and for that client the outcome is unknown rather than failed.

That is the part that sits outside the protocol. The consensus layer guarantees the log; the application still needs correct transaction semantics and client retries. A client that times out against a leader must not assume the write did not happen, and must retry with an identifier the state machine can deduplicate, as described in Idempotency, with a bounded and jittered retry schedule as in Retry & Backoff. Reads also need care: a leader that has been deposed but does not know it can serve a stale read. Implementations avoid that either by confirming leadership with a majority before answering (a read-index style check) or by holding a time-bounded lease, and the lease variant depends on bounded clock drift. If you build on top of a leader, a stale-leader write that reaches a separate resource, such as an external lock holder, is protected only by fencing tokens, discussed in Distributed Locks.

Majority arithmetic and where the cost goes

A majority of n nodes is floor(n/2)+1, so a group of 3 tolerates 1 failure and a group of 5 tolerates 2. Going from 3 to 4 adds a node and no extra fault tolerance, since the majority becomes 3. Commit latency is set by the leader's round trip to the follower whose acknowledgment completes the majority, plus the follower's durable flush, so on a multi-region cluster geography dominates.

The figure uses made-up round-trip times, but the shape is real. In a 3-node group the leader needs one follower, so its commit latency is the distance to its nearest peer. In a 5-node group it needs two, so it is the distance to the second-nearest. Moving the leader to a region far from the others changes every write's latency, not only the leader's own clients'. Placement therefore has to be explicit: put the leader near the writers, spread nodes so that losing any one region still leaves a majority, and decide what the minority side does. A cluster of two nodes in one region and one in another stops accepting writes if the two-node region is lost or isolated, while three regions with one node each survive any single-region loss. The minority side of a partition must refuse writes, and the application should show a deliberate error rather than hang.

Failover without consensus: what an acknowledgment is worth

Many relational deployments use a single primary with asynchronous streaming replicas rather than a consensus group. Their acknowledgment means something weaker: the primary has committed locally, not that a second machine holds the data. If the primary is lost before shipping the record, promoting a replica discards an acknowledged write. Synchronous replication closes the gap by waiting for a standby to flush before answering, at the cost of a round trip on every commit and a policy question when the standby is unreachable: stall writes, or continue without protection.

# postgresql.conf on the primary (PostgreSQL 10+ for the ANY syntax).
# Commit waits until ANY 1 of the two named standbys has flushed the commit record to its WAL.
synchronous_standby_names = 'ANY 1 (replica_a, replica_b)'   # names are the standbys' application_name
synchronous_commit        = on        # 'on' waits for the standby WAL flush when sync standbys are configured

# Optional, per transaction: also wait until a standby has replayed the commit, so a read on that
# standby sees it. Costs more latency; has no effect for standbys not in the synchronous set.
#   SET LOCAL synchronous_commit = remote_apply;

# Other values: 'off' (acknowledge before the local WAL flush), 'local' (ignore standbys),
# 'remote_write' (standby has the record in the OS cache, not necessarily on disk).

Two further limits. A promoted standby leaves the old primary alive and possibly still reachable by some clients, so failover tooling has to fence it off, by revoking its network path, its storage access or its credentials, before directing writes to the new primary; otherwise two nodes accept writes. And replicas are not backups. A DELETE without a WHERE, a bad migration or a corrupted page is replicated faithfully to every copy, which is why recovery from those depends on point-in-time-recoverable backups, covered with the write-ahead log in Database Foundations.

Partitioning assigns each key to one shard so that capacity grows by adding shards. The shard key decides almost everything afterward. A hash of the key spreads load evenly but makes range scans touch every shard. A range partition on time keeps range scans local but sends all new writes to the newest shard. A geographic key helps data-residency rules and latency but is only as balanced as your users' distribution. Techniques that limit how many keys move when a node is added are covered in Consistent Hashing, and the operational side of resharding is in Scaling.

Two consequences belong in a distributed-data discussion. First, a hot key concentrates load on one shard no matter how many shards exist: one celebrity account, one popular product or one global counter. Splitting the key into several sub-keys, caching reads, or batching updates are the usual remedies, and each changes the consistency of the value it touches. Second, an operation that touches keys on two shards is a distributed transaction. Atomic commit across shards with two-phase commit is possible and expensive: it holds locks across network round trips and blocks if the coordinator fails at the wrong moment. Designing the data model so that the operations that must be atomic live in one partition is usually cheaper, and where a workflow genuinely spans services the alternative is a Saga with explicit compensation.

Whenever more than one replica can accept a write for the same key, concurrent updates are a normal event, and the system needs a rule that picks or combines them. The rule is a business decision disguised as a database setting, because it determines which user's intent is lost.

Last-write-wins and why wall clocks lose data

Last-write-wins (LWW) attaches a timestamp to each write and keeps the highest. It is simple and convergent, and it is silently lossy. If the timestamps come from wall clocks on different machines, any skew larger than the real gap between two writes reverses their order, and the write that really happened last is discarded even though both were acknowledged. Races between genuinely concurrent writes lose one of them by construction.

LWW is a fine choice when the value is a full overwrite, losing the occasional concurrent update is acceptable, and the ordering comes from a clock you can trust: presence, last-seen time, a cache entry. It is a poor choice for anything where each write carries information the others lack, such as editing different fields of the same profile, appending to a list, or changing a balance. Using a per-key version assigned by a single owner, or a hybrid logical clock (below), improves the ordering but does not turn concurrent writes into a sensible merge.

Version vectors: detect concurrency instead of guessing

A version vector stores, per replica, how many updates that replica coordinated for the key. Comparing two vectors tells you whether one version descends from the other or whether they are concurrent. A write carries the context it was based on, so a later write can supersede exactly what it saw. Concurrent versions are kept as siblings and returned to the application, which merges them. The cost is real: clients must pass context back, APIs return multiple values, vectors grow with the number of coordinating nodes and need pruning or dotted variants, and every consumer has to implement a merge.

// Version vectors: a map of replica id -> counter of updates coordinated by that replica.
function compare(a, b) {
  let aAhead = false, bAhead = false;
  for (const id of new Set([...Object.keys(a), ...Object.keys(b)])) {
    const x = a[id] ?? 0, y = b[id] ?? 0;
    if (x > y) aAhead = true;
    if (y > x) bAhead = true;
  }
  if (aAhead && bAhead) return 'concurrent';
  if (aAhead) return 'a-newer';
  if (bAhead) return 'b-newer';
  return 'equal';
}
const mergeVectors = (a, b) => Object.fromEntries(
  [...new Set([...Object.keys(a), ...Object.keys(b)])].map((id) => [id, Math.max(a[id] ?? 0, b[id] ?? 0)]));

console.log(compare({ A: 2, B: 1 }, { A: 1, B: 1 }));   // 'a-newer'    the second is an ancestor, discard it
console.log(compare({ A: 2, B: 1 }, { A: 1, B: 2 }));   // 'concurrent' keep both as siblings
console.log(mergeVectors({ A: 2, B: 1 }, { A: 1, B: 2 })); // { A: 2, B: 2 }  context for the merged write

CRDTs: convergence by construction, for specific types

A conflict-free replicated data type defines its merge operation so that it is commutative, associative and idempotent, which means replicas converge regardless of the order or duplication of updates. Counters (G-counter, PN-counter), sets (add-wins or remove-wins), registers and sequences for collaborative text have well-known designs. They remove the need for coordination for the operations they model. They do not make an arbitrary business invariant conflict-free. A counter converges to the sum of all increments and decrements; it cannot know that the sum must not go below zero.

// A PN-counter CRDT: one increment map and one decrement map, each merged by per-replica max.
// Merge is commutative, associative and idempotent, so replicas converge in any delivery order.
const mergeMax = (a, b) => Object.fromEntries(
  [...new Set([...Object.keys(a), ...Object.keys(b)])].map((id) => [id, Math.max(a[id] ?? 0, b[id] ?? 0)]));

class PNCounter {
  constructor(id) { this.id = id; this.inc = {}; this.dec = {}; }
  add(n) { this.inc[this.id] = (this.inc[this.id] ?? 0) + n; }
  sub(n) { this.dec[this.id] = (this.dec[this.id] ?? 0) + n; }
  merge(o) { this.inc = mergeMax(this.inc, o.inc); this.dec = mergeMax(this.dec, o.dec); }
  get value() {
    const sum = (m) => Object.values(m).reduce((t, v) => t + v, 0);
    return sum(this.inc) - sum(this.dec);
  }
}

const eu = new PNCounter('eu'), us = new PNCounter('us');
eu.add(1);                          // one seat left; replicated to both regions
us.merge(eu);
// Partition: each region checks its own copy of the counter, then sells.
if (eu.value >= 1) eu.sub(1);
if (us.value >= 1) us.sub(1);
// Partition heals: merge in both directions.
eu.merge(us); us.merge(eu);
console.log(eu.value, us.value);    // -1 -1   converged, and the invariant value >= 0 is broken

For inventory, seat counts, quotas and financial balances, the choices are not a better merge function but a different structure: give one owner the authority to decide, divide the quantity into per-region allocations so each region can act alone within its share (escrow), or coordinate through a transaction or consensus group that may reject writes during a partition. The last section builds that case out.

Wall clocks are useful for display, logging and approximate ordering. They cannot prove causality. Clocks on different machines drift, are stepped by synchronization, and may be corrected backwards, and a timestamp says nothing about whether one event could have influenced another. A local monotonic clock is the right tool for measuring durations on one machine, and says nothing about other machines.

A Lamport clock is a counter that each process increments on local events and sends, and sets to the larger of its own value and the incoming one, plus one, on receipt. It has a one-directional guarantee: if event a happened before event b, then L(a) < L(b). The reverse does not hold, so a smaller timestamp does not mean an earlier or causally related event. Breaking ties with a process id gives a total order consistent with causality, which is useful for deterministic merging, but it still does not tell you whether two events were concurrent. Vector clocks, which keep a counter per process, can tell: they identify exactly which events are causally ordered and which are concurrent, at the cost of metadata that grows with the number of participants.

A hybrid logical clock (HLC) pairs a physical-time component with a logical counter. It stays close to wall-clock time, so timestamps remain meaningful for humans, range queries and expiry, while the counter and the receive rule guarantee that an event is stamped after every event it has seen. The sketch below runs: a node whose wall clock is 60 ms behind still stamps its events after the message it received.

// Lamport clock: increment on local events and sends; on receive take max(local, remote) + 1.
class Lamport {
  constructor() { this.t = 0; }
  tick() { return ++this.t; }
  receive(remote) { this.t = Math.max(this.t, remote) + 1; return this.t; }
}
const p = new Lamport(), q = new Lamport();
const sent = p.tick();                 // 1: p sends a message stamped 1
q.tick(); q.tick();                    // q has two unrelated local events: 1, 2
console.log(q.receive(sent));          // 3: the receive is ordered after both the send and q's earlier events

// Hybrid logical clock: l follows the largest physical time seen, c orders events within the same l.
class HLC {
  constructor(now = () => Date.now()) { this.now = now; this.l = 0; this.c = 0; }
  tick() {
    const pt = this.now();
    if (pt > this.l) { this.l = pt; this.c = 0; } else { this.c += 1; }
    return [this.l, this.c];
  }
  receive([ml, mc]) {
    const pt = this.now();
    const l = Math.max(this.l, ml, pt);
    if (l === this.l && l === ml) this.c = Math.max(this.c, mc) + 1;
    else if (l === this.l) this.c += 1;
    else if (l === ml) this.c = mc + 1;
    else this.c = 0;
    this.l = l;
    return [this.l, this.c];
  }
}
const fast = new HLC(() => 1000), slow = new HLC(() => 940);   // slow's wall clock is 60 ms behind
const msg = fast.tick();                                       // [1000, 0]
console.log(msg, slow.receive(msg), slow.tick());              // [1000,0] [1000,1] [1000,2]
// slow's events are stamped after the message it received even though its wall clock is behind,
// and l stays within the skew of real time instead of drifting like a pure counter.

Neither clock solves consistency by itself. They give you timestamps that are consistent with causality; they do not make two concurrent writes mergeable, and they do not let a replica know it has seen everything it needs. Some systems go further by exposing bounded clock uncertainty from specialized time hardware and making transactions wait out that uncertainty before committing, trading latency for external ordering. That is an infrastructure-specific design, not something to assume from ordinary NTP.

The practical question is what ordering the product requires, and where. Global order, where every event in the system is ordered relative to every other, needs a single sequencer or a consensus group and limits throughput and latency to what that component can do. Per-key order, enough for most entities, is cheap because each key has one leader or log partition, but gives no relation between different keys. Per-session order, which is what session guarantees need, is the cheapest and can be enforced by tokens at the edge. State which one each operation relies on, and pay for no more.

Two regions accept the last available seat while they cannot talk to each other. When the partition heals, an eventual merge produces a history containing two confirmed purchases for one seat, as the PN-counter example above showed in miniature. No convergence rule can make both valid, because the invariant (a seat has at most one owner) spans both regions and neither could see the other's decision. Availability and invariant preservation have to be decided together, as a product choice. There are three honest options.

• A single authoritative owner. All sales for a flight go to one region or one consensus group. Correctness is simple. A customer far from the owner pays the latency, and a region cut off from the owner cannot sell. This is the right default when sales volume per flight is modest.
• Bounded escrow allocations. Divide the inventory into per-region shares. A region sells from its own share with no cross-region call and cannot oversell, because it is the only writer for those seats. The price is stranded stock: seats unsold in one region are unavailable to another until rebalanced, so some customers are refused while seats exist elsewhere.
• Coordination that may reject writes. A transaction through a consensus group, or a quorum-based protocol, preserves the invariant globally and rejects or delays writes on the minority side during a partition.

A fourth is a business policy rather than an engineering one: deliberately allow bounded overbooking and compensate the losers, which is how the pattern in a Saga frames undo. That is a legitimate choice when the cost of a compensation is lower than the lost sales, and it should be made by product and finance, not discovered in production.

The calculator shows the trade the escrow option makes. With demand of 120% of the seats and an uneven demand split, a region with a small allocation sells out while the other leaves seats unsold, so some requests are refused although a seat exists. The independent-counter row shows the alternative: each region sells as though it owned everything, and the merge reveals how many seats were sold twice. Tune the allocation to expected demand per region, rebalance as sales progress, and keep a small pool under central control for the end of the sale.

-- PostgreSQL. Per-region seat escrow: each region may only sell seats it owns.
CREATE TABLE seat_escrow (
  flight_id text    NOT NULL,
  region    text    NOT NULL,
  allocated integer NOT NULL CHECK (allocated >= 0),
  sold      integer NOT NULL DEFAULT 0,
  PRIMARY KEY (flight_id, region),
  CHECK (sold <= allocated)             -- the database refuses to oversell the local allocation
);
CREATE TABLE seat_transfer (            -- donor side: one row per rebalancing grant
  transfer_id text PRIMARY KEY,
  flight_id   text    NOT NULL,
  from_region text    NOT NULL,
  to_region   text    NOT NULL,
  seats       integer NOT NULL CHECK (seats > 0)
);
CREATE TABLE seat_transfer_applied (    -- receiver side: makes applying a grant idempotent
  transfer_id text PRIMARY KEY
);

INSERT INTO seat_escrow (flight_id, region, allocated) VALUES ('FL100', 'eu', 60), ('FL100', 'us', 40);

-- Sell one seat in the EU region. Runs entirely in the local region, no cross-region call.
UPDATE seat_escrow
   SET sold = sold + 1
 WHERE flight_id = 'FL100' AND region = 'eu' AND sold < allocated;
-- 1 row updated: sold. 0 rows: this region's escrow is exhausted. Refuse, or ask for a transfer.

-- Donor: move 10 unsold seats away from EU. Debit first, so a lost grant strands seats but never oversells.
BEGIN;
INSERT INTO seat_transfer (transfer_id, flight_id, from_region, to_region, seats)
VALUES ('t-77', 'FL100', 'eu', 'us', 10)
ON CONFLICT (transfer_id) DO NOTHING;
-- 0 rows inserted: a retry of a grant that was already recorded. ROLLBACK and stop.
UPDATE seat_escrow SET allocated = allocated - 10 WHERE flight_id = 'FL100' AND region = 'eu';
-- If EU has fewer than 10 unsold seats the CHECK (sold <= allocated) raises an error and the
-- whole transaction, including the transfer row, is discarded.
COMMIT;

-- Receiver: apply the grant exactly once, even if the message is delivered several times.
BEGIN;
INSERT INTO seat_transfer_applied (transfer_id) VALUES ('t-77') ON CONFLICT DO NOTHING;
-- 0 rows inserted: already applied. ROLLBACK and stop.
UPDATE seat_escrow SET allocated = allocated + 10 WHERE flight_id = 'FL100' AND region = 'us';
COMMIT;

The sale path is a single guarded update in the local region, relying on the table constraint instead of check-then-act in application code. Rebalancing moves a seat count between regions, which is a two-step process across a failure boundary: the donor records the grant and reduces its allocation in one local transaction, and the receiver applies it idempotently once the grant arrives. Delivering the grant reliably from the donor's transaction is the Outbox Pattern, and applying it once under at-least-once delivery relies on the same idempotency key discipline as Idempotency.

Start with the operation, not the database. For each write path, name the invariant it must preserve and where that invariant is enforced. A unique constraint or a guarded update on a single primary enforces it in the database; an escrow allocation enforces it by ownership; a consensus-backed transaction enforces it by agreement; a merge function enforces nothing beyond convergence. Then name the freshness each read needs: must it see this session's writes, any committed write, or just something recent? Operations that fail the invariant test need a leader, ownership or consensus. Operations that only need convergence, such as counters of views or last-seen timestamps, can use the cheaper model. Reads that need session-level freshness get tokens; reads that only need to be roughly current can use replicas.

Then instrument the assumptions. Replication lag in time and bytes per replica, with alerts on the percentile your product tolerates, shows whether read-your-writes strategies are working. Quorum timeouts, hinted-handoff backlog and read-repair rate show how close a leaderless cluster is to inconsistency. Leader election count and term changes show instability in a consensus group, and commit latency by region shows what a multi-region placement actually costs. Clock offset between nodes matters because LWW, leases and HLC bounds all depend on it. The number of sibling or conflict resolutions per day tells you whether concurrent writes are rare or routine. For escrow, the unsold stock stranded per region and the rate of refused requests while seats exist elsewhere say whether the allocation matches demand.

Finally, test failure rather than assuming it: partition a replica set in a staging environment, kill a leader mid-write, skew a node's clock, and check that client-visible behavior matches the policy you documented, including what the minority side returns. When overload rather than partition is the trigger, the same systems fail in other ways, and Circuit Breaker and Load Shedding cover how to keep the failure from spreading. Read-heavy projections of the same data into search or analytics stores are the subject of Specialized Data Stores, and separating the write model from those read models is the focus of CQRS.