Consistency Models and Distributed Databases Questions

Direct answer

Eventual consistency means that after a write stops happening, all replicas of the data will eventually converge on the same value, but there is no guarantee about how long that takes or what a reader sees in the meantime. It trades a temporary window of staleness for lower write latency and higher availability, and it is the right default for data where a slightly-stale read is harmless, and the wrong default where a stale read causes real damage.

Structured elaboration

Whether eventual consistency is acceptable comes down to one question: what does the application actually do with a stale read?

• Tolerant workloads: anything the user does not act on financially or safety-critically in the moment. Order history, delivery-tracking analytics, recommendation feeds, and dashboard counters are all fine to serve slightly stale, because a few seconds of lag has no real consequence.
• Intolerant workloads: anything where a stale read causes an incorrect real-world action. Capturing a payment, decrementing the last unit of inventory, or checking an account balance before a withdrawal are all cases where a stale read can produce double-charges, oversells, or overdrafts.

The dividing line is not the technology, it is the cost of being wrong for a few hundred milliseconds to a few seconds.

Worked example

Picture a food-delivery app.

• Acceptable: the "your driver is 4 stops away" tracker and the "orders this month" analytics dashboard read from an asynchronously-replicated read replica. If that replica is a second behind, the customer sees the driver's position update a second late, which nobody notices.
• Not acceptable: the moment a customer taps "place order" and their card is charged. If two replicas of the payment-capture record briefly disagree about whether the charge already happened, a naive retry can charge the card twice. This path needs a strongly-consistent read (or an idempotency key tied to the order, so a retry is safe regardless of replication lag).

A second, different domain shows the same trade-off with a different shape of consequence. Picture a social-feed app instead: a user posts a photo and immediately likes their own post. Because "post visible to followers" and "like count" are two independently-replicated pieces of data, a reader can briefly see a user-visible anomaly: the poster's own like counted in the total but the post itself not yet visible in a follower's feed, or the reverse, the post visible but the like count still showing the pre-like value. Nobody's money or safety is at stake here, so full strong consistency for every post and every counter would be a wildly expensive fix for a cosmetic problem. The mitigation is much cheaper than moving to strong consistency everywhere: have the poster's own client apply an optimistic local update (show "liked", show the post as posted, immediately, from the write they just issued) regardless of what the shared aggregate view currently shows, while everyone else's feed is allowed to catch up asynchronously over the next second or two. This is the same "read-your-writes for the writer only" idea as the food-delivery payment case, just applied to a cosmetic anomaly instead of a financial one, which is the point: the fix pattern generalizes across very different domains and severities.

Trade-offs and mitigations

You rarely need to make the whole system strongly consistent to fix this. Options, cheapest first:

1. Read-your-writes for the writer only: route the customer's own immediate post-order reads (or, in the social-feed case, the poster's own view of their own post) to the primary or a replica guaranteed to have applied their write, while everyone else's dashboard or feed keeps reading from a lagging replica.
2. Idempotency keys on the write path itself, so even if a client retries under uncertainty, the payment is captured at most once regardless of what any read shows.
3. Reserve strong consistency for the specific field that matters (payment status, inventory count for the last few units) rather than promoting the entire order record, or the entire social graph, to strong consistency, which would slow down the majority of reads that never needed it.

The common mistake is treating "eventual consistency" as a single global switch. In practice it is a per-field decision: most of an application, whether it is a checkout flow or a social feed, can tolerate staleness, and only the handful of fields tied to money, safety, or the acting user's own immediate perception of their own action need the latency cost of strong consistency.

Direct answer

To give serializable transactions across continents at high throughput, the core design choice is how you order transactions globally without a single bottleneck node: Google Spanner does it with synchronized, bounded-uncertainty clocks (TrueTime) plus two-phase commit across Paxos-replicated groups of shards, while Calvin (the design behind FaunaDB) does it by deterministically pre-ordering every transaction through a sequencing layer before execution, avoiding cross-shard commit coordination entirely. I would build on the Spanner-style approach here, because it lets most transactions that touch a single shard commit without cross-region coordination, and only pay the cross-region cost for transactions that genuinely span shards.

Structured elaboration

• Global ordering / consensus: partition the data into shards, and replicate each shard across regions using a consensus protocol (Paxos or Raft) so each shard has a consistent, agreed-upon log even if some replicas are unreachable. A transaction that touches multiple shards additionally needs a commit protocol (below) to agree on a single global order across those shards.
• Clocks / timestamps across regions: assign every transaction a timestamp from a clock service with a known, bounded uncertainty window (Spanner's TrueTime uses GPS and atomic clocks to keep that window in the single-digit milliseconds). A transaction only commits once it has waited out that uncertainty window, guaranteeing that any transaction which starts after another one committed is assigned a strictly later timestamp, which is what makes the system's ordering externally consistent rather than merely internally consistent.
• Commit protocol: for a transaction spanning multiple shards, run two-phase commit across the Paxos leaders of the shards involved: each shard leader prepares (votes to commit), and once all have voted yes the coordinator commits everywhere. Single-shard transactions skip this entirely and commit through that shard's own consensus group, which is why sharding data so that most transactions touch one shard matters enormously for throughput.
• Reaching the 100,000-transactions-per-second target: the commit protocol operates per shard group, not globally, so aggregate throughput scales with the number of independent shard groups doing useful work in parallel, not with any single bottleneck. As a rough illustrative estimate, not a measured figure: if a single Paxos-replicated shard group, bound by local disk fsync and intra-region replication round trips, can sustain on the order of a few thousand single-shard commits per second, reaching 100,000 TPS aggregate is a matter of running on the order of dozens of shard groups concurrently, as long as the workload's transactions are mostly single-shard. This reframes "100,000 TPS" from a property of the commit protocol into a property of the sharding strategy: the commit protocol has to be correct and reasonably fast per shard, but the target throughput is met by parallelism across shards, not by making any single shard's consensus round faster.
• Measuring and mitigating cross-region latency: track the fraction of transactions that are single-shard versus multi-shard (multi-shard transactions pay the two-phase-commit round trip across regions and dominate tail latency), and the actual observed clock-uncertainty window (a wider TrueTime bound directly adds to every transaction's commit latency, since the transaction must wait it out). Mitigation is mostly a data-modeling problem: choose shard keys so that transactions which need to be atomic together (like a single customer's order) usually land on one shard, keeping the expensive multi-shard path rare rather than the default.

Worked example

For a transaction touching two shards in different regions with a 10ms round-trip and a 7ms clock-uncertainty bound, the commit path pays roughly the round-trip for two-phase-commit's prepare/commit round plus the uncertainty-wait, since the transaction cannot be certified as committed until both have happened. That is why the single biggest lever on throughput is not the commit protocol itself but the sharding strategy: if 95% of transactions are single-shard, only the remaining 5% pay this cost, and the system's effective throughput looks close to a single-shard system's, with a smaller tail of multi-shard transactions carrying the added latency.

Trade-offs and pitfalls

The alternative worth naming explicitly: Calvin-style deterministic pre-ordering avoids needing tightly-synchronized clocks or two-phase commit at all, by having every replica execute the exact same, pre-agreed sequence of transactions, which makes replicas trivially consistent by construction. Its cost is a sequencing layer that must batch and order all incoming transactions before any of them execute, adding a different kind of latency (batching delay) and requiring the full transaction's read/write set to be knowable up front, which is awkward for transactions that need to read data to decide what to write next.

A different family of solutions avoids cross-region database consensus for a transaction altogether by giving up atomicity at the database layer and pushing coordination into the application: an application-level saga. Instead of one atomic cross-shard commit, a saga breaks the operation into a sequence of local, single-shard transactions, each with its own compensating action if a later step fails. This buys lower per-step latency (no step waits on a cross-region 2PC round or a TrueTime uncertainty wait) at the cost of giving up atomicity and isolation across the sequence: intermediate state is visible to other readers before the whole sequence finishes, and a partial failure requires running compensations rather than a single rollback. The saga's own mechanics (how compensating transactions are structured, how partial-failure recovery works) are a distinct competency from this design question; the comparison that matters here is where the cost is paid: up front as commit latency inside the database (the Spanner-style design above), or later as compensation logic and temporarily-visible partial state in the application (a saga). For a workload that genuinely needs cross-shard atomicity as a database-level guarantee (the ask in this question), the synchronized-clock-plus-2PC approach is the right fit; a saga is the right fit when the application is willing to model the operation as a sequence of independently-committable steps instead.

The common mistake is treating "serializable global transactions" as one solved problem: the real decision is which of these mechanisms (synchronized clocks plus 2PC, deterministic pre-ordering, or pushing atomicity into an application-level saga) fits your workload's transaction shape and your tolerance for the specific kind of latency, or complexity, each one adds.

Direct answer

Once you already know two replicas disagree (read-repair and anti-entropy are how you detect that in the first place), the resolution toolkit has three pieces, and each answers a different question about what to do next. Vector clocks tell you whether two writes are genuinely concurrent or one causally follows the other, but do not by themselves pick a winner; last-write-wins (LWW) picks a winner by timestamp and silently discards the loser; CRDTs (Conflict-free Replicated Data Types) are data structures designed so that any two divergent replicas merge back into the same value automatically, with no discarded data and no coordination required.

Structured elaboration

• Vector clocks: each replica keeps a per-node counter, incremented on every local write. Comparing two vector clocks tells you if one is causally "before" the other (safe to just keep the later one) or "concurrent" (a true conflict). Vector clocks are a detection mechanism, not a resolution policy: they tell you a conflict exists, not what to do about it. A common resolution policy layered on top (the one Amazon's original Dynamo system uses) is to keep both concurrent versions as sibling values, return them together on the next read, and let the client or an application-level merge function decide; the shopping-cart use case merges siblings by unioning their items rather than picking one and discarding the other. The trade-off: if a client instead applies a naive "just pick one" policy (an implicit last-write-wins on top of vector-clock-detected concurrency), the discarded sibling's write is lost with no record it ever existed, which is exactly the failure the sibling-return approach is designed to avoid, at the cost of pushing the merge decision onto every client that reads a multi-valued key. Use case: detecting when a client's write was based on stale data. Pitfall at scale: the clock grows with the number of nodes/clients that have ever written the key, and for use cases with many writers (like collaborative editing) this metadata can grow faster than the data itself.
• Last-write-wins (LWW): on a detected conflict, keep whichever write has the later timestamp and discard the other. Use case: high-volume, low-stakes fields where an occasional lost update is acceptable (a "last seen" timestamp, a UI preference). Pitfall: it silently drops a genuinely concurrent write with no record it ever existed, which is unacceptable for anything where losing an update has real cost (an inventory decrement, a cart item).
• CRDTs: data structures (counters, sets, sequences) with a merge operation mathematically guaranteed to converge to the same result regardless of the order replicas merge in, with no coordination and no lost updates. Use case: counters, collaborative documents, presence/membership sets. Pitfall: not every data type has a natural CRDT formulation (a bank balance with a hard floor of zero is much harder to express as a CRDT than a simple counter), and CRDT metadata (tombstones for removed set elements, per-node counters) accumulates over time and needs periodic garbage collection or it silently grows the storage footprint.

Worked example (executed)

A grow-only counter (G-Counter) CRDT, where each node tracks its own increments and merge takes the max per node:

class GCounter:
    def __init__(self, node_id, node_ids):
        self.node_id = node_id
        self.counts = {n: 0 for n in node_ids}

    def increment(self, amount=1):
        self.counts[self.node_id] += amount

    def merge(self, other):
        merged = GCounter(self.node_id, self.counts.keys())
        for n in self.counts:
            merged.counts[n] = max(self.counts[n], other.counts[n])
        return merged

    def value(self):
        return sum(self.counts.values())

nodes = ["A", "B", "C"]
a = GCounter("A", nodes); a.increment(5)
b = GCounter("B", nodes); b.increment(3)
c = GCounter("C", nodes); c.increment(2)

print(a.merge(b).merge(c).value())  # A-then-B-then-C
print(c.merge(b).merge(a).value())  # C-then-B-then-A

Three replicas increment concurrently with no coordination (A does 5, B does 3, C does 2). Merging in order A-then-B-then-C gives value 10; merging in the reverse order C-then-B-then-A also gives value 10 (executed; both orders confirmed to converge to the same result). Extending to a PN-Counter (adds a second counter for decrements) lets the same technique represent both increments and decrements:

class PNCounter:
    def __init__(self, node_id, node_ids):
        self.node_id = node_id
        self.p = {n: 0 for n in node_ids}
        self.n = {n: 0 for n in node_ids}

    def increment(self, amount=1):
        self.p[self.node_id] += amount

    def decrement(self, amount=1):
        self.n[self.node_id] += amount

    def merge(self, other):
        merged = PNCounter(self.node_id, self.p.keys())
        for k in self.p:
            merged.p[k] = max(self.p[k], other.p[k])
            merged.n[k] = max(self.n[k], other.n[k])
        return merged

    def value(self):
        return sum(self.p.values()) - sum(self.n.values())

x = PNCounter("X", ["X", "Y"]); x.increment(10)
y = PNCounter("Y", ["X", "Y"]); y.decrement(4)
print(x.merge(y).value())

A concurrent +10 on one replica and -4 on another merges to 6 (executed; confirmed), with neither operation lost.

Contrast that with LWW on the same kind of concurrent update: two replicas write qty=8 and qty=3 a millisecond apart, with neither replica having seen the other's write. LWW keeps only the later timestamp and discards the earlier one with no record a conflicting write ever existed, which is exactly the failure mode a CRDT (or the sibling-return vector-clock policy above) is designed to avoid, at the cost of needing a merge-friendly data structure in the first place.

Trade-offs and pitfalls

For a reporting database that must maintain referential integrity across regions (a foreign-key-style reference from one table to another), none of these three techniques resolve the problem by themselves: LWW and CRDTs converge each row independently, but a reference can still point at a row that a concurrent delete removed on the other side. That needs an explicit dangling-reference detection pass layered on top of whichever base reconciliation technique you use, plus a policy for what to do with an orphaned reference (restore it, tombstone the referencing row too, or flag it for manual cleanup), and the reconciliation process needs to tolerate a temporarily-incompatible schema if one region has rolled forward a schema change the other has not yet applied. The common mistake is picking one technique (usually LWW, because it is the simplest to implement) for everything, rather than matching the technique to what a lost or merged update actually costs for that specific field.

Direct answer

The two teams' requirements are not actually incompatible on the SAME data, they are only in conflict if you insist on serving both from one undifferentiated read/write path. The fix is to split the responsibility: Team A's strongly-consistent writes stay strongly consistent (accept the latency), while Team B's reads are served from a fast, eventually-consistent read path, most simply a cache or read replica in front of the same source of truth, so Team B never touches the write path's latency at all.

Structured elaboration

• Keep the write path strongly consistent for Team A: writes go through a quorum (or a single leader) so the moment a write is acknowledged, it is durable and correct. This does not change; Team A's requirement is non-negotiable and it lives entirely on the write side.
• Decouple Team B's reads onto a fast path: introduce a cache, a read replica, or a materialized view that Team B reads from, refreshed asynchronously from the strongly-consistent source. This gets Team B's sub-50ms target because a cache read never pays the write path's coordination cost, at the cost of Team B's reads being slightly stale relative to the absolute latest write.
• Make the staleness bound explicit, not implicit: agree with Team B on how stale "sub-50ms reads" is allowed to be (a bounded-staleness guarantee, not unbounded eventual consistency), and monitor the actual replication lag against that bound so a degradation is caught before it silently exceeds what Team B's use case can tolerate.
• API design matters: expose the two access patterns as genuinely different API calls or endpoints (a strongly-consistent "read after my own write" endpoint for Team A's own confirmation flow, and a fast "read the latest available" endpoint for Team B), rather than one endpoint with an ambiguous consistency guarantee that neither team can rely on confidently.

Worked example

A checkout service (Team A) needs its own write, and its own immediate confirmation read, to be strongly consistent, since telling a customer "order placed" before the write is durable risks losing the order. A separate order-status widget elsewhere in the product (Team B, sub-50ms reads) does not need that guarantee: it reads from a cache populated asynchronously from the same underlying order data, refreshed within, say, a 200ms bound. Team A's write path and Team B's read path can share the same source of truth without either team paying for the other's requirement.

Trade-offs and pitfalls

The main operational cost of this split is now having two paths to keep healthy instead of one: the cache or replica feeding Team B's fast reads needs its own monitoring (is it actually staying within the agreed staleness bound, and what happens if replication falls behind), and a bug in cache invalidation can silently violate Team B's freshness assumption without anyone noticing until a customer complains. The common mistake is trying to satisfy both teams with a single tunable "consistency level" setting on one shared path, which forces a compromise that under-serves both (either Team A's writes get riskier, or Team B's reads get slower) instead of giving each team exactly what they actually need.

Direct answer

The core idea is that consistency becomes a parameter on each request rather than a fixed property of the whole database: the service stores each key with N replicas, and a client picks linearizable, causal, or eventual per key (or even per call) by choosing how many replicas the coordinator must contact and what ordering metadata it must check before returning. Linearizable calls pay a quorum round-trip and a version check; causal calls pay a smaller round-trip to confirm the client's own prior writes are visible; eventual calls pay almost nothing, a single nearby replica answers immediately.

Structured elaboration

• Components: a stateless coordinator layer that receives requests and fans them out to the replica set for a key; a replica store (each replica holds the key's value plus version metadata); a client library that lets the caller specify the desired consistency level per request and hides the fan-out/quorum logic behind a simple get(key, consistency) / put(key, value, consistency) API.
• Client behavior per tier:
  • Linearizable: the coordinator contacts a write quorum on put and a read quorum on get, sized so that R + W > N, guaranteeing every linearizable read sees the latest linearizable write.
  • Causal: the client library attaches a small causal-context token (a version vector covering the keys that client has previously read or written) to each request; the coordinator ensures the read reflects at least that context before returning, without requiring a full quorum.
  • Eventual: the coordinator returns whatever the nearest reachable replica has, no version check, no quorum.
• Failure handling: if a linearizable request cannot reach a quorum (say, during a partition), it must fail loudly rather than silently downgrading to a weaker guarantee the caller did not ask for; a causal or eventual request degrades more gracefully, since a caller who already accepted weaker guarantees can tolerate serving from fewer replicas.
• Routing and replication: replicate asynchronously to all N regions for eventual/causal reads to stay fast everywhere; for linearizable operations, route the quorum check to whichever replicas are needed to satisfy R or W, which usually means at least one round trip leaves the client's local region if the quorum cannot be satisfied locally.
• Meeting the 10,000-requests-per-second-per-region target: the coordinator layer is stateless, so it scales horizontally behind a load balancer, independent of the consistency tier a given request uses. The eventual and causal tiers carry almost none of the quorum-coordination cost, so most of the traffic (whichever tiers the workload's key mix favors) never becomes a bottleneck; the linearizable tier is the one that needs headroom planning, since every linearizable call consumes capacity on W replicas at once, so the replica set's write capacity, not the coordinator, becomes the limiting factor if a large share of traffic requests the strongest tier.

Worked example (executed quorum arithmetic for the linearizable tier)

For N=5 replicas and a fixed read quorum R=2, what is the minimum write quorum W that still guarantees linearizable reads, and what happens to that guarantee if W is set one lower?

def strongly_consistent(N, R, W):
    return (R + W) > N
min_W = N - R + 1  # = 5 - 2 + 1 = 4

Running this: with N=5, R=2, the minimum W is 4 (executed; confirmed). At W=4, strongly_consistent(5, 2, 4) returns True; at W=3 (one less), it returns False (executed; both confirmed). That single-replica difference is the entire gap between a linearizable guarantee and none: at W=4 the read quorum (2) and write quorum (4) are guaranteed to overlap in at least one replica, so any read is guaranteed to see the latest write; at W=3, R+W=5 equals N rather than exceeding it, so a read quorum and a write quorum can land on entirely disjoint sets of replicas and a client can read a stale value. Under a single node failure, this linearizable configuration (W=4 of 5) can still accept writes (4 of the remaining 4 healthy replicas can still form the quorum only if exactly one has failed; a second simultaneous failure would block writes entirely), which is the concrete availability cost of choosing the strongest tier.

Trade-offs and pitfalls

The main pitfall is letting a caller request "linearizable" and silently getting something weaker during a partition, which is worse than an honest failure, because the caller has no way to know their guarantee was violated. The API must make that distinction explicit (an error, not a downgraded success) for the linearizable tier specifically, while causal and eventual tiers can be designed to degrade gracefully since the caller already accepted a weaker guarantee going in. A second pitfall is ergonomics: if choosing the wrong tier for a given field is easy to do by accident (say, a default that quietly falls back to eventual), the client library should make the strongest-needed tier the explicit default for anything touching money or safety, and require an opt-in to relax it, rather than the reverse.