Replication, Partitioning, Sharding, and Quorum Systems - Complete Notes

A language-neutral guide to copying and distributing data, choosing quorum behavior, moving partitions safely, generating identifiers, and reasoning about correctness during replica, network, and routing failures.

00. The map, copies, and evidence mental model

Partitioning decides where a record belongs, replication decides how many failure domains hold a copy, and a quorum decides how much evidence an operation needs before it can succeed.

Imagine a library with several buildings. A catalog rule sends books A to F to one building and G to L to another. That is partitioning. Keeping three copies of an important book in separate buildings is replication. Requiring two librarians to confirm an update before accepting it is a quorum rule. The catalog, copy policy, and confirmation rule solve different problems and can fail independently.

Three decisions on every request
order_id = "ord-4821"
        |
        v
partition function or partition map
        |
        v
logical partition P17
        |
        v
replica placement: zone-a/node-4, zone-b/node-8, zone-c/node-2
        |
        v
coordination rule: leader ack, all acks, or W of N acks

Routing answers "where?"
Replication answers "which copies?"
Acknowledgement answers "how much evidence is enough?"
Replication is not a backup

Replicas quickly copy valid writes, accidental deletes, corruption, and malicious changes. A backup preserves an independently restorable historical state. Production systems commonly need replication for availability and backups for recovery.

01. Precise terminology and prerequisites

Clear terms prevent architecture discussions from mixing data layout, consistency, and physical deployment.

Replica
A maintained copy of some logical data, with a defined update and recovery protocol.
Replication factor, N
The intended number of replicas for a logical item or partition.
Leader, primary, or source
The replica authorized to order writes for a replication group.
Follower, secondary, or standby
A replica that applies changes ordered by a leader. Whether it may serve reads is a policy.
Replication log
An ordered record of changes or state transitions that another replica can replay. It may be a physical byte-level log or a logical record-level log.
Replication lag
The distance between a source and replica. Measure it in log positions, bytes, operations, and wall-clock age because any single measure can hide a different problem.
Partition
A logical subset of data selected by a deterministic placement rule.
Shard
Commonly, a partition placed on an independently scalable storage group. Product terminology varies, so state whether "shard" means a logical range, replica group, or server.
Horizontal partitioning
Splitting rows or records, usually by a partition key.
Vertical partitioning
Splitting columns, features, or ownership boundaries into separately stored groups.
Partition key or shard key
The value used to select a logical partition. It need not be the record's unique ID.
Partition map
A versioned mapping from key ranges or tokens to replica groups.
Coordinator
The node that receives a request and gathers enough replica responses for the chosen policy.
Quorum
A required subset of participants. In quorum replication, R is the read response count and W is the write acknowledgement count out of N replicas.
Failure domain
Infrastructure likely to fail together, such as one disk, host, rack, zone, region, power feed, network path, or administrative account.
Hotspot and skew
A hotspot is a placement receiving disproportionate work. Skew is uneven data size, request rate, or request cost across placements.
A successful write is defined by the protocol

An acknowledgement might mean buffered in memory, appended to an operating-system cache, flushed to durable media, persisted on one node, or persisted in several failure domains. Ask exactly what an acknowledgement proves before making durability claims.

02. Problems being solved and invariants

Replication and sharding are tools for explicit requirements, not automatic upgrades.

Requirement Likely technique New cost or risk
Survive a machine or zone loss Replicas across independent failure domains Write coordination, lag, failover, extra storage
Scale read throughput Read replicas or independently readable replicas Staleness, consistency routing, replica saturation
Exceed one node's storage or write capacity Horizontal sharding Routing, cross-shard operations, rebalancing
Place data near users or satisfy residency rules Region-aware replication and placement Wide-area latency, partitions, policy complexity
Keep independent data lifecycles or access controls Vertical partitioning Joins and atomic updates may cross boundaries

Before selecting a topology, write the invariants in observable terms:

  • An acknowledged paid order must survive loss of any one zone.
  • Two successful requests must never allocate the same public order ID.
  • A user must read their own confirmed message within two seconds.
  • A tenant's records must remain in its approved region.
  • A partition move must not lose or expose records to the wrong tenant.

Some invariants require consensus, transactions, uniqueness constraints, or single-writer ownership. Replication by itself does not create those guarantees. If two replicas accept conflicting inventory decrements without coordination, having more copies preserves the conflict rather than preventing overselling.

03. Leader-follower replication

A leader establishes one write order; followers replay that order to maintain copies.

Typical write path
Client          Leader              Follower A          Follower B
  | PUT x=9        |                     |                   |
  |--------------->| append log at 842  |                   |
  |                 |------------------->| apply 842         |
  |                 |--------------------------------------->| apply 842
  |                 |<-------------------| ack               |
  | success         |  acknowledgement policy satisfied     |
  |<----------------|                     |                   |

Replication log and apply pipeline

  1. The leader validates a command against its current state and assigns an order or log position.
  2. The leader appends the change to its log. Durability depends on the configured flush point.
  3. Followers fetch or receive log records, persist them, and apply them in order.
  4. The leader acknowledges after its configured local and remote conditions are satisfied.
  5. Followers expose applied state to reads only under their serving policy.

Physical replication replays storage-level changes and is usually tightly coupled to an engine version and layout. Logical replication sends record-level operations and supports filtering or transformation more easily, but it must preserve schema compatibility and transaction semantics. A snapshot or base copy supplies the starting state; the log carries changes after the snapshot's checkpoint.

Synchronous and asynchronous acknowledgement

Mode Success usually waits for Benefit Main trade-off
Asynchronous Leader persistence only Lower latency and remote failure isolation A failover can lose acknowledged but unreplicated writes
Synchronous receive Remote receipt or log persistence Better failure tolerance Does not always mean remote apply or read visibility
Synchronous apply Remote apply to queryable state Stronger immediate read behavior Write latency includes slow follower apply
Geographic synchronous Another region or fault domain Low data-loss objective across regional loss Wide-area round trips and reduced write availability

Read replicas and session correctness

Read replicas scale workloads that tolerate their lag, such as product browsing or analytics. They are unsafe as an invisible replacement for the leader when the next request must observe a just-confirmed write. Three practical read-your-writes strategies are:

  • Route the user's bounded session to the leader after a write.
  • Return a commit position and wait until a chosen replica has applied at least that position.
  • Use a consistency-aware read API that selects a sufficiently current replica or rejects.

Lag can come from network delay, a slow disk, long transactions, schema work, apply conflicts, insufficient CPU, or a follower serving too many queries. A follower that is healthy at the process level can still be unfit for freshness-sensitive traffic.

Failover state machine

Promotion with an explicit epoch
LEADER_ACTIVE(epoch 41)
        |
        | failure detector suspects leader
        v
ELECTION_OR_OPERATOR_DECISION
        |
        | choose sufficiently current candidate, obtain epoch 42
        v
NEW_LEADER_FENCED(epoch 42)
        |
        | publish routing, reject epoch 41 writes
        v
RECOVER_FOLLOWERS_AND_RECONCILE_OLD_LEADER

Failure detection is suspicion, not proof. A paused or partitioned old leader may still accept writes. Safe promotion therefore needs a mechanism such as consensus membership, an epoch, a fencing token, or storage-level exclusivity that makes the old writer unable to commit. Promotion policy must consider replay position, missing acknowledged writes, recovery time, and whether the candidate belongs to an independent failure domain.

Failover can trade availability for data loss

Automatically promoting an asynchronous follower may meet a recovery-time target while violating a zero-data-loss claim. Record the old leader's last durable position, the promoted position, and any divergent writes so operators can reconcile rather than silently discard evidence.

04. Multi-leader replication

Multiple leaders improve local write availability, but concurrent writers can create legitimate versions that no timestamp can safely interpret by itself.

Multi-leader topologies appear in multi-region systems, intermittently connected clients, and migration bridges. Each leader accepts local writes, then exchanges them with peers. They are useful when applications can name a deterministic merge or constrain each entity to one home writer. They are dangerous when business invariants require a single global order.

Concurrent profile edits
Initial: {email: old@example.com, phone: 111}

Region A, disconnected: set email = new@example.com
Region B, disconnected: set phone = 222

Whole-record last-write-wins may discard one independent edit.
Field-aware merge can preserve both because fields do not conflict.

But two concurrent changes to the same shipping address require
business resolution, not merely a larger timestamp.

Conflict detection

A system needs metadata that distinguishes causally newer versions from concurrent versions. Version vectors, per-record generations, operation identifiers, and hybrid logical metadata are examples. Wall-clock timestamps alone are vulnerable to skew, clock rollback, coarse resolution, and a malicious or misconfigured writer.

Merge strategies

  • Avoid: assign each entity a home leader, or serialize invariant-sensitive operations.
  • Last-write-wins: simple convergence, but can silently lose valid concurrent work.
  • Field merge: useful when independently changed fields have stable semantics.
  • Set union: useful for additions, but removals need tombstones or richer metadata.
  • Application merge: retain siblings and apply domain rules or request human review.
  • Convergent data type: use a proven CRDT whose operations have required algebraic properties.

Conflict resolution must be deterministic, associative where order can vary, idempotent under replay, and explicit about deletes. Test the merge of A with B in both orders, duplicate delivery, three-way concurrency, and a delete racing with an update.

05. Leaderless replication

A coordinator sends operations directly to several replicas and reconciles their possibly different versions.

Leaderless write with N = 3 and W = 2
Client -> Coordinator
              |---- write v8 ----> Replica A: ack
              |---- write v8 ----> Replica B: unavailable
              |---- write v8 ----> Replica C: ack
              |
              +---- two durable acknowledgements -> success

Replica B needs a hint, read repair, or anti-entropy later.

Leaderless does not mean coordination-free. The coordinator still chooses replicas, gathers acknowledgements, handles timeouts, compares version metadata, and may repair stale copies. It avoids one permanent write leader, but moves conflict and convergence work into every request and background maintenance.

Read repair

A coordinator reads version metadata or digests from replicas. If responses disagree, it fetches enough full values to choose or merge the correct result, returns according to the consistency policy, and updates stale replicas. Blocking repair improves the replicas involved but adds tail latency. Asynchronous repair shortens the client path but leaves a longer inconsistency window.

Anti-entropy repair

Background anti-entropy compares replicas independently of client reads. Hierarchical hash trees can locate differing ranges without transferring the full dataset. Matching root hashes avoid deep comparison; mismatching branches are recursively narrowed, then differing records are streamed. Anti-entropy is essential for cold keys that normal reads never repair.

Hinted handoff

When a target replica is temporarily unavailable, a coordinator can store a durable hint containing the missed mutation and destination. It replays the hint after recovery. Hints reduce the stale window, but retention can expire, the coordinator can fail, and replay can overload the returning node. Hinted handoff is a best-effort bridge, not a replacement for scheduled repair.

Tombstones are part of convergence

In an eventually replicated store, deleting a value usually creates a versioned tombstone. If the tombstone is discarded before every replica has learned it, an old value can return during repair. Tombstone retention, maximum outage, repair cadence, and replacement procedures must be designed together.

06. Quorum reads, writes, and their limits

Quorum arithmetic describes intersecting replica sets; it does not automatically provide a complete consistency guarantee.

For N replicas, a common rule is:

Intersection conditions
R + W > N     every read set intersects every completed write set
W > N / 2     any two write sets intersect

Example N = 3:
R=2, W=2      balanced quorum
R=1, W=3      fast reads, writes require every replica
R=3, W=1      fast writes, reads require every replica

Intersection gives a reader access to at least one replica that accepted a completed write only when the same replica membership is used, responses represent durable versions, and conflict metadata correctly identifies the latest version. A coordinator must actually inspect enough versions. Blindly returning the fastest response defeats the argument.

Tunable consistency

Operation policy Availability Latency Suitable example
ONE High while any suitable replica responds Usually low Approximate feed counters where staleness is acceptable
QUORUM Tolerates a minority of failures Waits for a majority and reconciliation User-visible profile state with bounded conflict handling
ALL Any unavailable replica can block Bound by slowest replica Rare operations where every replica must observe before success
LOCAL_QUORUM Survives local minority failure Avoids wide-area path Region-local requests with asynchronous cross-region convergence

Sloppy quorums

A strict quorum uses the designated N replicas. A sloppy quorum can accept writes on healthy fallback nodes outside that preference list when designated replicas are unreachable, often with hints for later handoff. This improves write availability, but read and write sets might not overlap during a partition. Saying "R plus W exceeds N" is then insufficient without defining the replica set, fallback behavior, and repair path.

Why quorum is not automatically linearizable

  • Two writers may concurrently create versions unless writes use a serialization protocol.
  • Last-write-wins can select a clock-skewed older business value.
  • A timed-out write may have reached a quorum even though the client sees failure.
  • Membership changes can create disjoint quorums if configurations are changed unsafely.
  • Sloppy placement can remove the expected read-write intersection.
  • Deletes and expired conflict metadata can resurrect state.

Linearizable registers normally need a protocol that orders writes and makes reads discover the latest completed order, such as consensus, a correctly implemented quorum register with versioned write-back, or a leader protected by epochs. Quorum count is one ingredient.

Failure tolerance

With N = 2f + 1 and majority reads or writes, up to f unavailable replicas can be tolerated for that operation. This assumes failures do not take a shared rack or zone containing most replicas. Replica count without independent placement gives false confidence. A three-replica set with two copies in one failed zone might have only one surviving copy.

07. Replica placement and topology

Copies should fail independently while remaining close enough to meet latency and cost targets.

A placement policy should state:

  • how many replicas belong to each partition;
  • which host, rack, zone, and region combinations are forbidden;
  • whether reads prefer local replicas and when they may cross regions;
  • how much failure capacity remains during maintenance;
  • which residency, encryption-key, and tenant policies constrain placement;
  • how replacement and decommission preserve temporary over-replication.
Placement for N = 3 across three zones
Region east
  zone a: P17 replica on node a4
  zone b: P17 replica on node b8
  zone c: P17 replica on node c2

Maintenance rule:
  never drain another P17 replica while one is unavailable

Capacity rule:
  surviving two zones must handle P17 read, write, repair, and failover load

08. Horizontal and vertical partitioning

Partitioning makes a subset independently manageable, but every split introduces operations that no longer fit inside one boundary.

Horizontal partitioning

Rows are divided by a key, such as tenant_id, conversation_id, or a hash of order_id. It scales storage and request throughput when common operations use that key. It makes cross-partition joins, constraints, sorting, and transactions more expensive.

Vertical partitioning

Columns or domains are separated. A user profile might keep public display fields in one store, authentication secrets in a highly restricted service, and large avatars in object storage. This can reduce row width, isolate sensitive data, and give components independent lifecycles. It also turns previously local reads and atomic updates into multi-system workflows.

Split Good fit Weakness
Tenant Isolation, residency, tenant-local queries One large tenant can dominate a shard
Entity ID hash Even point reads and writes Range queries scatter
Time range Retention, recent-window queries, archival Current range becomes a write hotspot
Feature or columns Security and independent scaling Cross-boundary reads and consistency

09. Range, hash, and directory partitioning

Range partitioning

A partition owns a contiguous interval such as dates 2026-07-01 through 2026-07-31 or customer IDs 1000 through 1999. Range scans and pruning are efficient. Boundaries can follow retention or geography. Monotonic keys direct new writes to the highest range, however, creating a hot partition unless ranges are split or the leading key spreads traffic.

Hash partitioning

Fixed-bucket placement
bucket = stable_hash(partition_key) mod bucket_count
partition = partition_map[bucket]

Requirements:
  stable_hash is identical across clients and versions
  partition_map has an explicit version
  bucket_count changes only through a migration protocol

Hashing usually spreads independent keys more evenly, but destroys natural order. A query lacking the partition key may contact every partition. Hashing does not fix a single hot key because every request for that key still selects one logical partition.

Directory or lookup partitioning

A metadata service maps a key or tenant to a shard. It allows intentional placement, fast tenant moves, and exceptions for large tenants. The directory becomes critical infrastructure, so cache entries need versions, updates need atomic publication, and stale routers need redirect or retry behavior.

10. Consistent hashing and virtual nodes

Consistent hashing limits how many key ranges move when membership changes; it does not remove the need for an authoritative map or safe data transfer.

Hash both keys and node tokens into a circular token space. Moving clockwise from a key token finds an owner; following owners can provide replicas. Adding a node takes selected token ranges instead of recomputing every key under a new modulo. Removing a node hands its ranges to successors.

Token ring sketch
                  token 0
                    |
             N4 ----+---- N1
            /                 \
         key K                   key M
            \                 /
             N3 ----------- N2

owner(K) = first assigned token clockwise from hash(K)
replicas(K) = next distinct eligible failure-domain owners

One token per physical node can produce uneven ownership and large movements. Virtual nodes assign many smaller token ranges to each physical node. They smooth capacity differences and stream from many peers during replacement, but increase metadata, repair pairings, and the number of ranges affected by one machine failure. Modern systems may instead use many fixed logical partitions and place those partitions through a central or consensus-backed map.

11. Routing and versioned partition maps

Data can be correct on storage nodes and still appear unavailable because routers disagree about ownership.

Router behavior with stale metadata
function route(request):
    map = local_partition_map
    target = map.owner(hash(request.partition_key))
    response = target.send(request, map.version)

    if response is STALE_MAP(new_version, hint):
        refresh_map_at_least(new_version)
        retry_once_with_same_idempotency_key(request)

    return response

Common routing models are client-side routing, a stateless proxy, or forwarding by any storage node. Client routing saves a hop but spreads metadata logic into every client. A proxy centralizes it but needs independent scaling. Forwarding simplifies clients but consumes storage-node network capacity and may add unpredictable hops.

Maps need monotonically increasing versions or epochs. A destination should reject a request that would write under an obsolete ownership epoch, and it should return enough metadata for the router to refresh. Unbounded redirect loops indicate metadata convergence failure and must be surfaced.

12. Secondary indexes, fan-out, and cross-shard work

Local secondary indexes

Each shard indexes only its own records. Writes remain local, but a query by an attribute that does not include the shard key must scatter to shards, merge results, and enforce a global limit or sort. Tail latency approaches the slowest required shard, and partial failures need explicit semantics.

Global secondary indexes

A global index maps an alternate key, such as customer_email, to record IDs or shards. It avoids scatter but is itself partitioned and replicated. Synchronous index maintenance makes writes more expensive and may require distributed transactions. Asynchronous maintenance permits stale or missing entries and needs reconciliation.

Safe scatter-gather

  • Set an overall deadline and per-shard concurrency limit.
  • Push filters and local top-K limits down to each shard.
  • Merge using stable sort keys and a deterministic tie-breaker.
  • Use continuation tokens containing map version and per-shard cursors.
  • Define whether missing shards produce an error, partial result, or stale fallback.
  • Prevent one user query from fanning out into unbounded internal requests.
Offset pagination breaks across moving shards

Independent writes and moves change shard ordering between pages. Prefer cursor pagination with an immutable sort tuple, such as (created_at, order_id), and state what consistency snapshot the cursor represents.

13. Hotspots, skew, and tenant placement

Equal bytes do not imply equal work, and equal request counts do not imply equal cost.

Measure at least these distributions per logical partition and physical node:

  • stored bytes, live rows, tombstones, and compaction work;
  • reads, writes, scans, fan-out participation, and bytes transferred;
  • CPU time, disk I/O, cache hit ratio, queue depth, and tail latency;
  • replication lag, repair backlog, hint volume, and migration traffic;
  • top keys and tenants, using privacy-safe sampling and bounded cardinality.

Corrections by hotspot type

Cause Safer correction Trade-off
Monotonic leading key Hash prefix, time buckets with sub-buckets, or randomized ID Range reads must merge buckets
Single celebrity or room key Read cache, replicated materialization, or split derived data Invalidation and merge complexity
Large tenant Dedicated placement or sub-shard within tenant Special routing and operational policy
Expensive query class Admission control, precomputation, or workload isolation Freshness or feature limits
Unequal node capacity Weighted placement and capacity-aware balancing More complex failure calculations

Salting a hot key must preserve a way to find all salts. For write-heavy counters, choose a fixed shard count, update one shard, then sum shards for reads. This raises read cost and only supports invariants that tolerate distributed counter semantics.

14. Rebalancing, resharding, and online migration

Moving ownership is a consistency protocol with bulk data transfer, not merely a file copy.

A safe range-migration state machine

Move range R from shard A to shard B
1. PREPARE(epoch 70)
   B allocates capacity; A remains authoritative.

2. SNAPSHOT_AND_COPY
   Copy a consistent snapshot of R to B; record source log position L.

3. CATCH_UP
   Stream changes after L. Verify checksums and counts by subrange.

4. DUAL_APPLY_OR_FORWARD(epoch 70)
   Keep B current while A still serves authoritative traffic.

5. CUTOVER(epoch 71)
   Atomically publish B as owner. A rejects epoch 70 writes and redirects.

6. OBSERVE_AND_REPAIR
   Compare source and destination; monitor errors, lag, and stale routing.

7. CLEANUP_AFTER_GRACE
   Delete A's old copy only after rollback window, backups, and map convergence.

Alternative protocols briefly pause writes at cutover or use change-data capture rather than dual writes. A raw application dual write is unsafe because one destination can succeed and the other fail. The migration controller needs durable progress, idempotent steps, retries, and a single ownership epoch.

Production controls

  • Throttle migration below disk, network, compaction, and replica-repair headroom.
  • Limit concurrent moves that share a source, destination, rack, or network path.
  • Pause automatically when customer latency, error rate, or replication lag breaches limits.
  • Validate counts, hashes, sampled records, tombstones, and secondary indexes.
  • Keep old data read-only until the rollback and stale-router window closes.
  • Make cancellation resume-safe; never infer completion only from a missing worker process.

Resharding can amplify I/O because copied data competes with foreground traffic, replication, compaction, and backup. Plan temporary free space for source, destination, logs, and retained old copies. Test rebalancing while a node fails, not only on an idle healthy cluster.

15. Globally unique identifier strategies

An identifier can provide uniqueness, locality, or approximate order, but each additional meaning changes its failure and privacy properties.

Strategy Strength Failure or cost Good fit
Single database sequence Simple uniqueness and total allocation order Central dependency, contention, visible volume One write authority with moderate allocation rate
Range or hi-lo allocation Clients allocate locally from reserved blocks Gaps, stranded ranges, allocation-service recovery Shards needing numeric IDs without per-ID coordination
UUIDv4 Decentralized, random 122-bit payload Large index, random locality, collision still probabilistic Public opaque IDs and independent writers
UUIDv7 Standard time-ordered layout with random bits Leaks creation time and needs rollback handling Rough time locality when timestamp exposure is acceptable
Snowflake-style integer Compact, decentralized, roughly time ordered Worker-ID coordination, clock rollback, topology leakage High-rate internal event or entity creation
Random token Unpredictability with sufficient entropy Longer representation and collision calculation Capability or externally enumerable resource IDs

Sequence block allocation

Hi-lo idea
allocator atomically reserves block 8301
worker can issue IDs 8301000 through 8301999 locally

crash after ID 8301042:
  unused IDs become gaps, but must never be reassigned

Uniqueness requires durable non-overlapping block ownership.
Gap-free numbering is a different, much more expensive requirement.

Snowflake-style bit layout

Illustrative layout, not a universal standard
| time since custom epoch | worker ID | per-tick sequence |

on generate:
  now = clock_millis()
  if now < last_time:
      reject, wait, or use a proved rollback strategy
  if now == last_time and sequence exhausted:
      wait for next tick with a deadline
  persist or safely retain last_time where restart can reuse worker ID

Worker IDs must be unique among simultaneously active generators. A lease without fencing can let an old process and replacement share one ID. Clock rollback can duplicate the same time-worker-sequence tuple. Never silently set a backward clock to the last seen value without proving the sequence space cannot repeat across restart.

UUID choices and ordering leakage

RFC 9562 defines UUID formats including random UUIDv4 and Unix-time-based UUIDv7. A UUID is an identifier, not an authorization secret unless it is generated with adequate unpredictability and protected as a capability. Time-ordered IDs can improve some index-locality patterns but reveal approximate creation time and may let observers estimate activity. Sequential IDs additionally enable enumeration and expose rough counts. Use an opaque public ID when those leaks matter.

16. End-to-end sharded order system

The following design is suitable for tenant-local order access with high write volume; it does not pretend that every cross-tenant query is cheap.

Requirements and assumptions

  • 40,000 writes per second at peak and 200,000 point reads per second.
  • Orders average 3 KiB including indexed metadata before replication.
  • Most customer and support reads include tenant_id and order_id.
  • A confirmed order must survive one zone loss; region disaster RPO is under one minute.
  • Per-order state transitions require optimistic version checks.
  • Cross-tenant analytics may be minutes stale and runs from a separate replicated pipeline.

First-pass capacity

Approximate daily storage and shard load
40,000 writes/s * 86,400 s/day * 3 KiB = about 9.9 TiB/day logical at peak all day

If measured peak-to-average ratio is 4:
  average raw growth is about 2.5 TiB/day

With replication factor 3:
  about 7.5 TiB/day before indexes, logs, compaction, and backups

With 128 logical partitions:
  ideal peak average per partition = 312.5 writes/s
  provision above p99 partition load, not ideal average

Keys and placement

  • Partition key: stable hash of (tenant_id, order_id) into 128 logical partitions.
  • Directory: versioned map from logical partition to one three-zone replica group.
  • Replication: one leader and two followers per group, synchronously durable in two zones.
  • ID: UUIDv7 or a fenced generator, plus tenant ID in authorization context, not inferred from ID.
  • Customer history: a separate tenant-bucketed index, maintained through a durable event log.
  • Analytics: asynchronous export, so broad scans do not overload transactional shards.
Create-order flow
Client
  | POST order with tenant credential and idempotency key
  v
API gateway -> authenticate, rate limit, bind tenant context
  v
Order router -> hash tenant_id + order_id, read map epoch 71
  v
P17 leader -> check idempotency and expected state
  |          append order + outbox in one local transaction
  |          replicate durably to second zone
  v
Client receives order ID, committed version, and partition epoch

Async indexer consumes outbox and updates customer-history view.
Reconciler checks source orders against derived index.

Replica failure behavior

If one follower fails, the leader and other follower can still satisfy the two-zone acknowledgement rule, but the group has no further fault margin. Alert on under-replication, reserve recovery bandwidth, and block planned maintenance for that group. Reads requiring read-your-writes use the leader or an applied-position check. Stale-tolerant reads can use followers within a lag budget.

Network partition behavior

A minority side cannot elect or remain an authorized leader. The majority side increments the group epoch and continues if its placement and durability policy are satisfied. The isolated old leader must be fenced from durable storage or reject writes after losing its authority lease. When connectivity returns, it rejoins as a follower from a known log point or receives a new snapshot.

Ambiguous client timeout

Timeout after commit
Client       Leader       Follower
  | request    |             |
  |----------->| commit v12  |
  |            |------------->| durable ack
  |            | success     |
  X response lost

Retry with same idempotency key:
  leader returns recorded result for v12

Retry with a new key:
  could create a duplicate order

Online partition split

If P17 is hot, split its hash interval into P17a and P17b rather than increasing the global modulo. Copy and catch up P17b to a new replica group, cut over with a new map epoch, and keep redirects on the source. Customer-history indexes store record IDs and use the current directory, avoiding a rewrite of every index entry when physical ownership changes.

17. Production deployment and operations

Deployment rules

  • Roll one replica per group at a time and prove quorum headroom before each drain.
  • Use readiness checks for log catch-up and serving eligibility, not process existence alone.
  • Keep replication protocol and on-disk format compatibility through mixed-version rollout.
  • Canary both storage nodes and routers because stale routing has different failure modes.
  • Freeze balancing during risky schema changes, incident response, or constrained capacity.
  • Restore backups into isolation and verify application-level records, not only file checksums.

Observability

Area Signals Diagnostic question
Replication lag bytes/time, apply rate, log retention, under-replicated groups Is data received but not applied, or not received?
Quorum responses by consistency level, unavailable, timeout, conflict rate Which replica or domain prevents enough evidence?
Distribution bytes, QPS, CPU, disk, p99 latency by partition and tenant Is imbalance caused by data, traffic, or request cost?
Repair hint age, replay rate, repair coverage, mismatched ranges Can replicas converge before tombstones expire?
Migration bytes copied, catch-up lag, checksum failures, stale-map redirects Is cutover safe and have routers converged?
IDs worker lease conflicts, clock rollback, sequence exhaustion, collisions Can two generators issue the same bit tuple?

Troubleshooting playbook

  1. Classify impact by tenants, partitions, regions, operations, and consistency levels.
  2. Capture map epoch, replication group, coordinator, and operation ID from a failing request.
  3. Compare leader log position with each replica's received, durable, and applied position.
  4. Check whether latency comes from quorum wait, disk queue, network, repair, or migration.
  5. Pause rebalancing and expensive repairs if they amplify foreground impact.
  6. Restore quorum by recovering the safest current member, not simply the fastest-to-start copy.
  7. Reconcile ambiguous writes through operation IDs and authoritative logs.
  8. After mitigation, validate convergence and restore fault margin before closing the incident.

18. Security, privacy, abuse, and trust boundaries

Replication multiplies sensitive copies, and routing metadata can become an authorization hazard.

  • Authenticate and mutually authorize replication peers. Network reachability alone must not grant permission to stream or replace data.
  • Encrypt transport and storage, rotate credentials, and scope keys by environment, region, or tenant where the threat model requires it.
  • Derive tenant authorization from verified identity, then compare it with the record. Never trust a user-provided shard or tenant routing hint as authorization.
  • Protect partition maps, repair commands, snapshot endpoints, and rebalancing controls as privileged control-plane operations with audited changes.
  • Apply deletion and retention to replicas, hints, logs, indexes, snapshots, backups, and migration leftovers. Track proof of completion under applicable policy.
  • Keep regional replicas and temporary migration copies inside residency boundaries. A short-lived copy is still a data transfer.
  • Rate-limit fan-out and expensive consistency levels. Attackers can turn one public request into work on every shard or replica.

Sequential and time-based IDs reveal information. Do not expose internal shard numbers, worker IDs, region codes, exact creation times, or business volume unless intended. An unguessable ID reduces enumeration but does not replace object-level authorization.

19. Performance, scalability, and cost

Write cost

A logical write may cause N replica writes, log I/O, secondary index writes, compaction, change capture, backup traffic, and later repair. The client latency is set by the required acknowledgement path, while total resource cost includes every asynchronous copy. Batching improves throughput but increases queueing delay and the amount retried after failure.

Tail latency

Waiting for all replicas makes latency track the slowest. Waiting for a quorum tracks an order statistic, such as the second-fastest of three, but only if stragglers are cancelled or safely ignored. Scatter-gather across many shards makes at least one slow response increasingly likely. Measure complete operation percentiles, not average per-node latency.

Failure and maintenance headroom

Size normal operation so a failed domain's traffic can move without saturating survivors. Include log catch-up, hint replay, repair, rebuild, and reshard traffic in the failure model. A cluster at 70 percent disk throughput may have no safe headroom when one of three zones disappears and the other two absorb its work.

Cost model

Useful first-order model
stored bytes = logical bytes * replication factor
             + indexes + retained logs + tombstones + migration overlap

network bytes = replication + repair + cross-region reads
              + backups + rebalancing + client traffic

operating cost = steady resources + failure headroom
               + control plane + on-call and migration complexity

20. Failure scenarios and safer responses

Scenario Unsafe reaction Safer response
Leader unreachable from one zone Promote a leader on every side Allow only an authorized quorum side and fence the old epoch
Follower lag grows Keep sending freshness-sensitive reads Remove from that read class, diagnose receive versus apply lag
Quorum write times out Retry with a new operation ID Retry idempotently or query outcome by original ID
Returning replica has old values Trust its process health Keep non-serving until catch-up and repair prove consistency
Hot partition Add replicas expecting writes to spread Split write ownership or change key design if semantics allow
Stale partition map Accept write at old owner Reject stale epoch, refresh, and retry idempotently
Migration interrupted Delete source because copy started Resume durable phase state; source remains authority until cutover
Clock moves backward on ID worker Continue with reset sequence Stop, wait safely, or use a proved logical-time fallback
Repair after tombstone expiry Merge old value as live Keep repair interval below retention and replace stale nodes safely

21. Common mistakes and corrections

  • Mistake: "Three replicas means no data loss." Correction: define acknowledgement durability, failure-domain placement, and failover position.
  • Mistake: treating read replicas as always current. Correction: expose lag and route by the operation's freshness requirement.
  • Mistake: last-write-wins by wall clock for payments. Correction: use idempotent state transitions, versions, and an authoritative order.
  • Mistake: quoting R + W greater than N as proof of strong consistency. Correction: state membership, versioning, read selection, concurrent-write, and repair behavior.
  • Mistake: sharding before measuring one-node limits. Correction: first optimize data model and indexes, then shard for demonstrated scale or isolation needs.
  • Mistake: choosing a shard key only for even storage. Correction: include request rate, request cost, locality, fan-out, and future growth.
  • Mistake: changing hash(key) mod N when adding a node. Correction: use stable logical partitions or consistent hashing plus controlled migration.
  • Mistake: deleting source data immediately after cutover. Correction: retain it through map convergence, verification, and rollback grace.
  • Mistake: assuming globally unique means globally ordered. Correction: define uniqueness and ordering separately.

22. Testing strategy

Unit and property tests

  • Partition function is deterministic across supported versions and languages.
  • Every key maps to exactly one active logical owner for a given epoch.
  • Replica placement never repeats a forbidden failure domain.
  • Conflict merge is deterministic, idempotent, and order-independent where required.
  • ID generator does not duplicate across sequence exhaustion, restart, and clock rollback.
  • Quorum calculator handles unavailable, delayed, duplicate, and stale responses.

Integration and compatibility tests

  • Bootstrap from snapshot plus log while writes continue.
  • Read-your-writes token waits for apply position and respects a deadline.
  • Mixed-version routers interpret map epochs and hash encodings identically.
  • Schema changes replay on old and new replicas through a rolling upgrade.
  • Secondary index reconciliation repairs lost, duplicate, and stale entries.
  • Tenant authorization remains enforced through redirects and forwarded requests.

Fault, partition, and recovery tests

  • Kill leader before local flush, after local flush, and after each remote acknowledgement.
  • Partition leader from clients, followers, or only one zone; verify exactly one write authority.
  • Delay one replica without failing it to reveal tail-latency and queue effects.
  • Expire hints, withhold reads, and prove anti-entropy repairs a cold key.
  • Crash every migration phase and prove restart reaches a safe state.
  • Restore an old snapshot and verify it cannot serve until log catch-up and fencing.

Load and capacity tests

  • Use realistic key popularity, tenant skew, read-write ratio, payloads, and fan-out.
  • Measure healthy load, one-zone load, rebuild load, and migration load.
  • Run a soak test long enough to expose compaction, log retention, and repair interaction.
  • Test hot-key limits separately because uniform generators conceal them.
  • Verify throttles protect foreground SLOs and that pausing migration actually reduces work.

23. Hands-on exercises and scenarios

Exercise 1: Quorum history

Model N = 3, R = 2, W = 2. Delay replica C, time out a write after A and C persist it, then read A and B. Explain why the client cannot assume the write failed and how version comparison returns the newest state. Expected reasoning: timeout is ambiguous; the read set intersects the persisted write set at A; idempotency prevents duplicate effects.

Exercise 2: Select a chat partition key

Compare user_id, conversation_id, and message_id. Expected reasoning: conversation ID preserves ordered room reads and participant fan-out, but a huge public room is hot. Add time or sequence sub-buckets only with an ordered merge plan.

Exercise 3: Failover drill

Disconnect the leader from two followers but leave it reachable to some clients. Expected reasoning: the isolated leader must lose authority, the majority chooses a new epoch, stale writes are rejected, and the old node returns through catch-up rather than immediate service.

Exercise 4: Online range move

Implement an in-memory source, destination, change log, and versioned router. Inject a crash at each migration phase. Expected result: each restart resumes idempotently, one epoch has one owner, and no acknowledged record disappears.

Exercise 5: ID risk review

Choose IDs for public orders and internal events. Expected reasoning: opaque random or suitable UUID for public enumeration resistance, time-oriented ID only when ordering and leakage are acceptable, and authorization independent of both.

Exercise 6: Capacity under zone loss

Given three zones at 45 percent CPU each, estimate surviving load when one zone fails and repair begins. Expected reasoning: remaining zones receive about 1.5 times foreground share before repair overhead, so validate CPU, disk, connection, and network headroom under the combined workload.

24. Interview questions and detailed model answers

1. What is the difference between replication and partitioning?

Partitioning divides the dataset so different subsets can scale or be managed independently. Replication copies a subset so it survives failures or serves more reads. A sharded database usually combines both: each shard owns part of the key space and each shard has several replicas.

Follow-up: Why can adding replicas fail to increase write throughput?

Every write may still pass through one leader and be copied to more nodes. Scaling write ownership generally requires additional partitions, while replicas add durability or read capacity.

2. Compare leader-follower, multi-leader, and leaderless replication.

Leader-follower simplifies write ordering but leader availability and failover matter. Multi-leader supports local writes in several sites but needs conflict avoidance or merge. Leaderless sends to multiple replicas and uses quorum plus reconciliation, improving some failure behavior while adding version and repair complexity. The workload's invariants and partition tolerance select among them.

Follow-up: Which would you choose for bank balances?

Prefer one serialized authority per account or a transactional consensus-backed design. Uncoordinated last-write-wins replicas are inappropriate for preserving a balance invariant.

3. What does synchronous replication guarantee?

Only what its acknowledgement point defines. It may wait for remote receipt, durable log flush, or apply. Ask how many replicas, which failure domains, and what happens when they are unavailable. Synchronous replication improves durability but increases latency and can reduce write availability.

4. Explain R + W greater than N.

It makes every read set intersect every completed write set under one stable designated replica set. The reader can then encounter at least one replica with the write. Correctness still depends on durable acknowledgements, version comparison, membership, concurrent writes, deletes, and repair. It is not a one-line proof of linearizability.

Follow-up: What changes with sloppy quorum?

Fallback nodes may not belong to the normal preference list, so read and write sets can be disjoint during a partition. Hinted handoff and repair restore convergence later.

5. What is replication lag and how do you measure it?

It is the distance from source state to a replica. Track received, durable, and applied log positions, byte backlog, estimated time lag, and apply throughput. Time-only lag can look small during no traffic; byte-only lag does not show user-visible age.

6. Why is last-write-wins risky?

"Last" is often selected by a wall clock that can skew or move backward. A later timestamp can overwrite a causally newer or independently valid update. It also hides conflicts instead of resolving their business meaning. Use causal metadata and domain merge rules when losing a version is unacceptable.

7. Compare range and hash partitioning.

Range partitioning preserves order and allows pruning but monotonic inserts can hotspot. Hash partitioning spreads keys but turns range queries into scatter-gather. Choose from access patterns, not only evenness, and state how rebalancing changes ownership.

8. What problem does consistent hashing solve?

It reduces remapping when membership changes compared with a direct modulo by node count. Virtual nodes improve distribution granularity. It does not solve hot keys, durability, safe transfer, membership agreement, or routing authorization.

9. How do you choose a shard key?

Start from dominant reads, writes, invariants, and locality. Estimate key cardinality, data and traffic skew, growth, hot tenants, range-query needs, fan-out, and move cost. A good key keeps common atomic work local and spreads the constrained resource. Validate it with production-like distributions.

10. How do secondary indexes work in a sharded system?

A local index exists on each shard and requires scatter for queries missing the shard key. A global index is another distributed mapping that avoids scatter but adds write consistency and recovery work. State whether updates are transactional or asynchronous and how stale entries are reconciled.

11. How do you move a shard online?

Copy a consistent snapshot, stream changes after its checkpoint, verify destination catch-up, then atomically publish a new ownership epoch. Reject stale writes, retain redirects, validate hashes, and delete the old copy only after a rollback grace period. Every phase needs durable idempotent progress.

12. What is read repair versus anti-entropy?

Read repair fixes differing replicas discovered on a client read, sometimes before responding. Anti-entropy scans and compares ranges in the background, so it covers cold data. Hints replay missed writes after short outages but are best effort. A robust eventually consistent system needs a repair plan independent of normal reads.

13. How would you handle one very large tenant?

Detect it before saturation, then use dedicated placement or sub-shard the tenant by another stable key. Preserve tenant-local authorization and query planning. Do not apply special routing invisibly without map versioning, capacity reservation, and a move-back procedure.

14. Compare sequence, UUIDv4, UUIDv7, and Snowflake-style IDs.

A sequence is simple and ordered but centralized. UUIDv4 is decentralized and random but wider and less index-local. UUIDv7 is standardized and time-oriented but leaks time and needs correct clock handling. Snowflake-style integers are compact and roughly ordered but require unique worker IDs, sequence limits, and clock-rollback safety. None substitutes for authorization.

15. How does a network partition affect the order design?

Only the side with authority for the latest epoch may accept writes. A minority refuses writes rather than create two leaders. The majority can continue if it still satisfies durability placement. Clients retry with idempotency keys because lost responses make outcomes ambiguous. Recovery fences and catches up the old side before serving.

16. When should you not shard?

Do not shard when one well-designed store meets measured capacity, reliability, and isolation needs. Sharding adds routing, rebalancing, distributed query, testing, and on-call costs. Vertical scaling, indexes, caching, archival, read replicas, or workload isolation may solve the actual constraint first.

25. Revision cheat sheet

  • Partitioning chooses the subset; replication preserves copies; quorum chooses evidence.
  • Replicas are not backups. Replication also copies mistakes.
  • An acknowledgement must name its memory, disk, apply, replica, and failure-domain boundary.
  • Leader-follower orders writes; multi-leader merges; leaderless reconciles replica versions.
  • Lag has received, durable, applied, byte, and time dimensions.
  • Failover needs authority epochs and fencing, not failure suspicion alone.
  • R + W greater than N gives intersection under assumptions, not automatic linearizability.
  • Sloppy quorum improves availability but can remove normal read-write intersection.
  • Read repair covers read keys; anti-entropy covers cold ranges; hints cover short outages.
  • Range partitioning helps scans; hash partitioning helps spread; directory maps help exceptions.
  • Consistent hashing limits remapping but does not move data safely by itself.
  • A hot key stays hot under hashing. Split its work or materialize reads when semantics allow.
  • Global indexes avoid scatter but create another consistency and repair problem.
  • Online moves need snapshot, catch-up, epoch cutover, verification, grace, and cleanup.
  • Unique, ordered, opaque, and unguessable are different ID properties.
  • Test zone loss together with repair and migration load, not only isolated node crashes.

26. Current primary official references

Last reviewed · July 2026 · part of knowledge-base