Design a multi-region active-active key-value store

"Design a key-value store replicated across five regions, where every region accepts writes."

Step 1: clarify, and force the consistency decision (5 minutes)

Active-active across regions means accepting concurrent writes to the same key in different regions, which means either coordinating (and paying a cross-region round trip on every write) or not coordinating (and resolving conflicts afterwards). There is no third option, and getting the interviewer to choose is the first job.

Cross-region round-trip latency (typical, real):
  us-east-1 <-> us-west-2      ~ 60 ms
  us-east-1 <-> eu-west-1      ~ 75 ms
  us-east-1 <-> ap-southeast-1 ~ 220 ms
  eu-west-1 <-> ap-southeast-1 ~ 180 ms

A quorum write across 5 regions must reach 3, so it waits for the
SECOND-slowest of the four remote regions.
From us-east-1 that is roughly 75 ms, before any processing.
From ap-southeast-1 it is roughly 180 ms.

So the question to ask:

"What is the write latency budget, and what is the consequence of two regions writing the same key concurrently? If the budget is single-digit milliseconds, I cannot coordinate, so I need conflict resolution and I need to know what a conflict costs you. If a conflict is unacceptable (a balance, an inventory count, a unique username), I have to coordinate and the budget must absorb a cross-region round trip."

Assume: single-digit-millisecond local writes, so asynchronous replication with conflict resolution, and assume the data model tolerates it. Then handle the cases that do not, in step 7, because a real system has both.

Other clarifications:

Scale        100 TB total, 500k reads/sec and 50k writes/sec globally
Value size   Average 2 KB, p99 64 KB
Access       ~85% of a user's traffic hits their home region
Durability   No acknowledged write may be lost after a single region
             is destroyed
Consistency  Read-your-own-writes within a region is REQUIRED.
             Cross-region convergence within seconds is acceptable.

That last line is the actual specification, and extracting it is most of the design work.

Step 2: capacity math (3 minutes)

Storage
  100 TB logical. Replicated to 5 regions = 500 TB.
  Within a region, 3 replicas for local durability = 1.5 PB raw.
  This is the cost of active-active and it should be said out loud:
  5x the storage bill before local replication.

Replication bandwidth
  50k writes/sec x 2 KB = 100 MB/sec of new data
  Each write ships to 4 other regions = 400 MB/sec cross-region egress
  = 34 TB/day of egress. At typical cloud egress pricing this is a
  five-figure monthly line item and it is often the thing that kills
  naive designs.
  -> Compression and batching are not optimisations here, they are
     required. Batched + compressed at ~4:1 -> ~100 MB/sec, ~8.6 TB/day.

Metadata for conflict resolution
  Version vectors: one entry per region per key = 5 entries x ~12 bytes
  = 60 bytes of metadata on a 2 KB value. 3% overhead. Acceptable.
  Compare with: a vector clock per CLIENT would be unbounded, which is
  why Dynamo-style systems version by node/region, not by client.

Read traffic
  500k reads/sec, 85% local. Local reads never cross a region, so
  each region serves ~100k reads/sec of its own traffic.
  At ~20k reads/sec/node, that is ~5 nodes per region minimum,
  more for storage capacity.

The number worth volunteering: 34 TB/day of cross-region egress. Candidates design active-active replication and never mention that the network bill is often larger than the compute bill, and batching plus compression is what makes it viable.

Step 3: architecture

   REGION A                REGION B               REGION C
 ┌──────────┐            ┌──────────┐           ┌──────────┐
 │ coordinat│            │ coordinat│           │ coordinat│
 │  ors     │            │  ors     │           │  ors     │
 ├──────────┤            ├──────────┤           ├──────────┤
 │ storage  │            │ storage  │           │ storage  │
 │ nodes    │            │ nodes    │           │ nodes    │
 │ (consist │            │ (consist │           │ (consist │
 │  hashing)│            │  hashing)│           │  hashing)│
 └────┬─────┘            └────┬─────┘           └────┬─────┘
      │                       │                      │
      └───────────┬───────────┴──────────┬───────────┘
                  ▼                      ▼
        ┌──────────────────────────────────────┐
        │  REPLICATION LOG (per region)        │
        │  ordered, durable, batched, compressed│
        └──────────────────────────────────────┘
                  │
        Each region tails every other region's log,
        applies with conflict resolution, tracks a
        per-source cursor for resumability.

Within a region: local quorum, strongly consistent. N=3, W=2, R=2, so W+R>N and a local read sees the latest local write. Latency is sub-millisecond because it is intra-AZ or cross-AZ within a region.

Across regions: asynchronous, eventually consistent. A write is acknowledged as soon as the local quorum commits, and replication happens in the background.

Why this split is the right one: it gives read-your-own-writes for the 85 percent of traffic that stays in its home region, which is the property users actually notice, at local latency. Cross-region convergence is measured in seconds, which nobody notices unless they move regions mid-session, and step 6 handles that.

Step 4: conflict resolution

Three mechanisms, and the design uses all three for different data.

Last-writer-wins with hybrid logical clocks

The default, and it is only defensible if you are honest about what it loses.

@dataclass(order=True)
class HLC:
    """Hybrid logical clock: physical time for human interpretability
    and bounded divergence, logical counter to break ties without
    depending on clock precision, node_id for total order."""
    physical_ms: int
    logical: int
    node_id: str

    @staticmethod
    def now(last: "HLC", node_id: str) -> "HLC":
        wall = int(time.time() * 1000)
        if wall > last.physical_ms:
            return HLC(wall, 0, node_id)
        # Physical clock did not advance (or went backwards):
        # advance the logical counter instead. This is what makes
        # the clock monotonic despite NTP corrections.
        return HLC(last.physical_ms, last.logical + 1, node_id)

What LWW loses, stated plainly: the losing write is gone. If region A sets cart = [X] and region B sets cart = [Y] concurrently, one of them silently vanishes and the user's item disappears. Plain wall-clock LWW is worse still, because clock skew between regions determines the winner, so a region whose NTP is 200 ms fast wins every conflict.

Hybrid logical clocks (Kulkarni et al., 2014) fix the skew problem: the order is monotonic and consistent with causality even when physical clocks disagree, and the physical component keeps timestamps human-meaningful. They do not fix the lost-write problem, and saying so is what separates a real answer from a memorised one.

Version vectors, to detect conflicts rather than hide them

# One counter per REGION, not per client. Bounded at 5 entries.
VersionVector = dict[str, int]     # {"us-east-1": 12, "eu-west-1": 7}

def compare(a: VersionVector, b: VersionVector) -> str:
    a_greater = any(a.get(k, 0) > b.get(k, 0) for k in a | b)
    b_greater = any(b.get(k, 0) > a.get(k, 0) for k in a | b)
    if a_greater and b_greater: return "CONCURRENT"   # a real conflict
    if a_greater:               return "A_DOMINATES"
    if b_greater:               return "B_DOMINATES"
    return "EQUAL"

Version vectors detect concurrency; they do not resolve it. On CONCURRENT you either return both siblings to the client (Dynamo's approach, which pushes the decision to the application that has the semantics) or apply a merge function.

Versioning per region rather than per client is the design decision, because a vector with one entry per client grows without bound, which is the known operational problem with naive vector clocks.

CRDTs, where the data model allows

class GCounter:
    """A grow-only counter. Merge is per-region max, which is
    commutative, associative and idempotent, so replicas converge
    regardless of message order or duplication."""
    def __init__(self): self.counts: dict[str, int] = {}
    def increment(self, region: str, n: int = 1):
        self.counts[region] = self.counts.get(region, 0) + n
    def value(self) -> int: return sum(self.counts.values())
    def merge(self, other: "GCounter") -> "GCounter":
        m = GCounter()
        m.counts = {r: max(self.counts.get(r, 0), other.counts.get(r, 0))
                    for r in self.counts | other.counts}
        return m

CRDTs eliminate conflicts by construction, which is qualitatively better than resolving them, and they cost expressiveness: an OR-Set, a G-Counter and a PN-Counter cover shopping carts, view counts and likes, and none of them can express "set this value to exactly X" or "only if the balance stays above zero".

The routing rule I would apply:

Counters, sets, flags, presence     -> CRDT. Converges, no loss.
User profile, settings, documents   -> LWW with HLC. Simple, and a
                                       lost concurrent edit is rare
                                       and tolerable.
Balances, inventory, uniqueness     -> NOT eventually consistent.
                                       These need coordination. See step 7.

Step 5: replication, anti-entropy and durability

1. Write commits to local quorum, appended to the region's
   replication log with its HLC and version vector.
2. A shipper batches log entries (say, 50 ms or 1 MB, whichever
   first), compresses, and streams to each peer region.
3. Each peer applies with conflict resolution and advances a
   per-source cursor, so a disconnect resumes rather than restarts.
4. Anti-entropy: periodic Merkle-tree exchange per key range
   catches anything the log missed.

Merkle trees are the mechanism worth explaining, because the naive alternative does not scale:

Compare a 20 TB key range between two regions by shipping all keys?
  Impossible.
Compare root hashes?
  One comparison. Equal -> done, nothing to do.
Unequal -> descend into the differing subtree only.
  Finding one divergent key in 20 TB is ~log(n) hash comparisons,
  so tens of exchanges rather than millions.

Durability against total region loss: the write is acknowledged on local quorum, which means a region destroyed one second after acknowledging loses up to one second of writes. If the requirement is genuinely "no acknowledged write may be lost", then that requirement contradicts single-digit-millisecond writes, and the honest answer is to say so and offer a per-key durability level:

DURABILITY_LOCAL     ack on local quorum.      ~2 ms.  Loses <1 s on
                                                        region loss.
DURABILITY_REGIONAL  ack after 1 remote region ~65 ms. Survives one
                     confirms.                          region loss.
DURABILITY_GLOBAL    ack after majority of      ~80 ms. Survives two.
                     regions confirm.

Offering a per-write durability level rather than a single system-wide answer is the staff-level move, because the requirement differs per key and forcing one answer means either paying 80 ms for session data or risking a payment record.

Step 6: reads, and the session guarantees

Read path (local): R=2 of N=3 within the region.
  Fast, and consistent with local writes.

The problem: a user whose request is routed to a different region
(mobile roaming, DNS change, region failover) can read a value
OLDER than their own write. That is the failure users actually
notice and complain about.

Session tokens are the fix, and they are cheap:

# The client carries the version it last observed. The serving
# region either satisfies it or waits briefly for replication.
def read(key: str, session_token: VersionVector | None) -> Value:
    local = store.get(key)
    if session_token and compare(session_token, local.vv) == "A_DOMINATES":
        # We are behind what this client already saw. Options:
        #   1. wait briefly for replication to catch up  (bounded)
        #   2. read from the region that has it          (slow, correct)
        #   3. return stale and mark the response        (fast, honest)
        waited = wait_for_version(key, session_token, timeout_ms=50)
        if not waited:
            return read_from_region(session_token.dominant_region(), key)
    return local

That gives monotonic reads and read-your-own-writes across region changes, which are the two session guarantees users perceive, without paying global coordination on every read. This is the mechanism most candidates omit, and it is what makes "eventual consistency" acceptable in practice.

Step 7: the data that cannot be eventually consistent

Every real active-active system has some. Pretending otherwise is the weakest part of most answers.

Unique usernames        Two regions can both accept "alice".
                        LWW picks one, and the other user has an
                        account that silently stops working.

Account balance         Two concurrent withdrawals of $80 from $100
                        both succeed locally. LWW loses one, so the
                        balance is wrong by $80 in the bank's favour
                        or the customer's. Either is a real incident.

Inventory               Overselling the last unit.

Idempotency keys        Two regions both process the "same" payment
                        because neither saw the other's key.

Three approaches, and I would use the first two:

1. Home-region ownership per key. Each key has a designated owner region, derived from the key or from the user's home. Writes are routed there; other regions forward. This gives single-writer semantics per key with no consensus protocol, at the cost of cross-region latency for the minority of writes that originate elsewhere. This is the cheapest correct answer and it covers most of these cases.

2. Consensus for the small set that genuinely needs it. A Raft or Paxos group spanning regions, for uniqueness constraints and balances. Writes cost a cross-region round trip (60 to 180 ms) and it is correct. Keep the data in it small, because everything in the consensus group pays that latency.

3. Reservation instead of decrement, which sidesteps the problem for inventory: partition the 100 units as 20 per region, so each region decrements locally with no coordination, and rebalance in the background when a region runs low. Correct, no coordination on the common path, and it can under-sell (one region has stock while another is sold out), which is usually far preferable to overselling.

Saying "here is the subset that cannot be eventually consistent, and here is how I handle it separately" is the strongest single move in this design, because the interviewer is usually probing for whether you will claim active-active solves everything.

Step 8: failure modes

Region isolated (network partition)
  -> It keeps accepting local writes (AP choice). Divergence
     accumulates, bounded by partition duration x write rate.
     On heal, conflict resolution runs over the backlog. The metric
     that matters is REPLICATION LAG per region pair, alerted.

Region destroyed
  -> Traffic fails over via DNS/anycast. Data survives in 4 regions.
     Up to ~1 s of DURABILITY_LOCAL writes lost, which was the
     documented trade rather than a surprise.

Clock skew / NTP correction
  -> HLC is monotonic by construction, so a backwards NTP step
     advances the logical counter instead of producing a timestamp
     that goes backwards. This is exactly why plain wall-clock LWW
     is not acceptable.

Replication log falls behind
  -> The shipper is a queue; if it backs up, either throttle writes
     (protects convergence, hurts availability) or let lag grow
     (protects availability, grows divergence). Make it a per-key-class
     policy, and alert on lag well before it becomes hours.

Conflict storm (a hot key written in all regions)
  -> Version vectors grow siblings without bound in a Dynamo-style
     design. Cap sibling count, and for a key showing repeated
     conflicts, promote it to home-region ownership automatically.

Split brain during failover
  -> Never allow two regions to believe they own the same key range.
     Ownership changes go through the consensus group, not through
     DNS, because DNS convergence is not a coordination primitive.

Step 9: what changes at ten times the scale

At 1 PB and 500k writes/sec:

Cross-region egress becomes the dominant cost, decisively. 500k writes/sec at 2 KB is 1 GB/sec to each of four peers, so 4 GB/sec of egress. The move is selective replication: not every key goes to every region. Replicate a user's data to their home region plus one backup, and fetch cross-region on the rare access. That converts a full mesh into a sparse one and cuts egress by most of it.

Version vector metadata stops being negligible for small values. At 500 byte values, 60 bytes of metadata is 12 percent overhead, and the move is to drop version vectors for key classes using CRDTs or home-region ownership, keeping them only where concurrent multi-region writes are genuinely expected.

Anti-entropy scheduling becomes a real problem. Merkle-tree exchange over a petabyte per region pair, per cycle, is expensive. The move is to scope it by recency: full trees only over recently-written ranges, with a slower background sweep of cold data.

Production evidence

DeCandia et al., "Dynamo: Amazon's Highly Available Key-value Store" (SOSP 2007) is the primary source for consistent hashing, version vectors with sibling reconciliation, Merkle-tree anti-entropy, and the explicit choice of availability over consistency. Its discussion of returning siblings to the application is the origin of the "detect, do not resolve" position.

Kulkarni et al., "Logical Physical Clocks and Consistent Snapshots in Globally Distributed Databases" (2014) defines hybrid logical clocks, which is what CockroachDB and several others use instead of wall clocks.

Shapiro et al., "A comprehensive study of Convergent and Commutative Replicated Data Types" (INRIA 2011) is the CRDT reference, including the proofs that merge must be commutative, associative and idempotent.

Riak implemented Dynamo's model in production including sibling resolution and later added CRDTs (Riak DT) precisely because application-level sibling merging proved to be the hardest operational burden, which is useful evidence for preferring CRDTs where the data model allows.

DynamoDB Global Tables uses last-writer-wins across regions and documents it explicitly, which is a good example of a large production system choosing the simple mechanism and being clear about what it loses.

Cassandra's multi-datacenter replication with LOCAL_QUORUM for local consistency and asynchronous cross-datacenter replication is the same architecture as step 3, and its tunable per-query consistency level is the per-write durability idea from step 5.

The debate

The case for active-active with async replication: local write latency, survival of any region failure without failover, and no coordination on the common path. For session data, carts, preferences, feeds and counters this is straightforwardly correct.

The case for single-writer with global reads: one region owns writes, others serve reads. No conflicts by construction, much simpler to reason about, and correct for balances and uniqueness. The cost is write latency for distant users and a failover process that has to be rehearsed.

The case for a globally-consistent database (Spanner, CockroachDB): correctness without conflict-resolution reasoning, at 10 to 100 ms write latency. If the budget allows it, it removes an entire category of application-level complexity, and the complexity it removes is the kind that produces subtle data-loss bugs years later.

My position: async active-active as the default, with an explicit per-key-class routing rule, and a small consensus group for the data that cannot tolerate it.

The routing rule is the design: CRDTs for counters, sets and flags, because they converge without loss; LWW with hybrid logical clocks for profiles and documents, because a lost concurrent edit is rare and tolerable; home-region ownership for anything with a uniqueness or monotonicity requirement; and a cross-region consensus group for balances and idempotency keys. One consistency model for the whole store is the mistake, and committing to the routing rule is what makes the design defensible.

Two things I would refuse. Plain wall-clock LWW, because clock skew then decides which write survives, and a region whose NTP is 200 ms fast wins every conflict silently. Hybrid logical clocks cost almost nothing and remove that failure entirely.

And claiming that acknowledged writes survive region loss while also promising single-digit-millisecond writes. Those are contradictory, and the honest answer is a per-write durability level so the caller chooses. Presenting that as a feature rather than as a caveat is the difference between an answer that sounds confident and one that is correct.

Where I would push back on the premise: most systems that ask for active-active want low read latency globally and low write latency locally, and do not actually need writes accepted everywhere for the same key. Home-region ownership with global read replicas delivers that with no conflict resolution at all, and it is worth proposing before building the harder thing.

Follow-up Q&A

"How do you resolve concurrent writes to the same key in two regions?" It depends on the data class and I would route by class rather than pick one mechanism. Counters, sets and flags go to CRDTs, which converge by construction with no loss. Profiles and documents go to last-writer-wins with hybrid logical clocks, accepting that a concurrent edit is lost, which is rare and tolerable there. Anything with a uniqueness or monotonicity requirement, balances and usernames, does not go to eventual consistency at all: it gets home-region ownership or a consensus group. One consistency model for the whole store is the mistake.

"Why hybrid logical clocks rather than timestamps?" Because with wall clocks, clock skew decides which write survives. A region whose NTP is 200 milliseconds fast wins every conflict, silently and permanently, and a backwards NTP correction can make a later write lose to an earlier one. Hybrid logical clocks keep the physical component for interpretability and advance a logical counter when the physical clock does not move, so the order is monotonic and consistent with causality regardless of skew. They cost about nothing and remove the failure entirely. What they do not fix is that LWW still loses a write, and that is a separate decision.

"What do version vectors actually give you?" Detection, not resolution. Comparing two vectors tells you whether one dominates or whether they are genuinely concurrent, which is the information you need to decide what to do. On concurrency you either return both siblings to the application, which has the semantics to merge them, or you apply a merge function. And the vector is per region, not per client, because a per-client vector grows without bound, which is the well-known operational problem with naive vector clocks.

"A user's balance is in this store. What happens?" It does not go in the eventually consistent path, and I would say so rather than try to make LWW work. Two concurrent withdrawals of eighty dollars from a hundred both succeed locally and LWW discards one, so the balance is wrong by eighty dollars. The options are home-region ownership, where the key has a designated owner and other regions forward writes to it, which gives single-writer semantics with no consensus protocol; or a cross-region Raft group, which costs a 60 to 180 millisecond round trip and is correct. I would use ownership by default and consensus only for the small set that needs stronger guarantees.

"You said no acknowledged write may be lost, and also single-digit-millisecond writes. Can you have both?" No, and that contradiction is worth surfacing rather than designing around silently. Acknowledging on local quorum means a region destroyed one second later loses up to a second of writes. Waiting for a remote region costs at least 60 milliseconds. So I would offer a per-write durability level: local quorum at about two milliseconds, one remote confirmation at about sixty-five, a global majority at about eighty. The caller chooses per key class, and session data and payment records get different answers.

"How do you detect divergence between regions?" Merkle trees over key ranges. Compare root hashes: equal means the ranges match and there is nothing to do, one comparison for a twenty-terabyte range. Unequal means descend into only the differing subtree, so finding a divergent key is logarithmic rather than linear. That is what makes anti-entropy affordable at all, and it is the mechanism behind Dynamo's and Cassandra's repair processes.

"A user moves between regions mid-session and sees old data. Fix it." Session tokens. The client carries the version vector it last observed, and the serving region compares. If it is behind what the client already saw, it waits briefly for replication, and if that times out it reads from the region that has the version. That gives read-your-own-writes and monotonic reads across region changes, which are the two guarantees users actually perceive, without paying global coordination on every read. Most candidates omit this and it is what makes eventual consistency acceptable in practice.

"What does this cost that people forget?" Cross-region egress. Fifty thousand writes a second at two kilobytes shipped to four peers is four hundred megabytes a second, which is thirty-four terabytes a day, and at cloud egress pricing that is a five-figure monthly line item. Batching and compression are not optimisations here, they are what makes the design viable, and at ten times the scale the answer becomes selective replication rather than a full mesh.

"When would you not build this?" When the requirement is really low read latency globally and low write latency locally, which is what most teams asking for active-active actually want. Home-region ownership with global read replicas delivers that with no conflict resolution at all, and it is a much simpler system to operate and reason about. I would propose it first and only build full active-active if writes genuinely have to be accepted everywhere for the same key.

Common misconceptions

"Active-active means no failover." It means no failover for writes to keys whose region is up. Key ownership, routing and session guarantees still need handling when a region goes away.

"Last-writer-wins is a resolution strategy." It is a discard strategy. One write is gone. That is sometimes fine and it must be a decision rather than a default.

"Vector clocks solve conflicts." They detect them. Resolution is a separate, application-level decision.

"CRDTs solve everything." They cover counters, sets and registers. They cannot express "set exactly X" or "only if the balance stays positive".

"Eventual consistency means users see stale data." With session tokens they do not see data older than what they already saw, which is the guarantee that actually matters.

Interview delivery note

Force the consistency decision in the first two minutes, with numbers, because the whole design branches on it: "Active-active means either coordinating on every write, which from us-east-1 across five regions is about 75 milliseconds for a quorum and 180 from ap-southeast-1, or not coordinating and resolving conflicts afterwards. There's no third option. So: what's the write latency budget, and what does a conflict cost you?"

Then commit to the routing rule rather than to one mechanism: "I wouldn't pick one consistency model for the whole store. CRDTs for counters, sets and flags, because they converge with no loss. Last-writer-wins with hybrid logical clocks for profiles and documents. Home-region ownership for anything with a uniqueness or monotonicity requirement. And a small consensus group for balances and idempotency keys. One model for everything is the mistake."

Volunteer the thing candidates claim and should not: "And I'd name the subset that cannot be eventually consistent, because every real active-active system has one. Two regions can both accept the username 'alice'; last-writer-wins picks one and the other user's account silently stops working. Two concurrent eighty-dollar withdrawals from a hundred both succeed locally and one is discarded."

The line that shows operational experience: "and I'd flag the cross-region egress early. Fifty thousand writes a second at two kilobytes to four peers is thirty-four terabytes a day, which at cloud pricing is often a bigger line item than the compute. Batching and compression aren't optimisations here, they're what makes it viable."

Close with session tokens, since it is the mechanism most people omit: "and I'd add session tokens so a user who moves regions never reads data older than what they already saw. That's read-your-own-writes and monotonic reads across region changes without global coordination, and it's what makes eventual consistency acceptable to actual users."

The Staff & Team Lead Interview Reference