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.
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?"
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.
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.
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
- The leader validates a command against its current state and assigns an order or log position.
- The leader appends the change to its log. Durability depends on the configured flush point.
- Followers fetch or receive log records, persist them, and apply them in order.
- The leader acknowledges after its configured local and remote conditions are satisfied.
- 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
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.
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.
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.
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.
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:
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.
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
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 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.
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.
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
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
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
| 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_idandorder_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
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.
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
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
- Classify impact by tenants, partitions, regions, operations, and consistency levels.
- Capture map epoch, replication group, coordinator, and operation ID from a failing request.
- Compare leader log position with each replica's received, durable, and applied position.
- Check whether latency comes from quorum wait, disk queue, network, repair, or migration.
- Pause rebalancing and expensive repairs if they amplify foreground impact.
- Restore quorum by recovering the safest current member, not simply the fastest-to-start copy.
- Reconcile ambiguous writes through operation IDs and authoritative logs.
- 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
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 Nwhen 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
- PostgreSQL documentation: Log-Shipping Standby Servers - streaming, synchronous replication, standby operation, and failover foundations.
- PostgreSQL documentation: Table Partitioning - range, list, hash partitioning, pruning, maintenance, and partition-count trade-offs.
- Apache Cassandra documentation: Dynamo architecture - consistent hashing, virtual nodes, replication, read repair, hints, and anti-entropy.
- Apache Cassandra documentation: Hints - hinted handoff lifecycle, retention, replay, and best-effort limitations.
- Apache Cassandra documentation: Repair - Merkle-tree comparison and full or incremental anti-entropy repair.
- MongoDB documentation: Distribute Collection Data - ranged and hashed sharding and distribution behavior.
- MongoDB documentation: Sharded Cluster Balancer - range migration, balancing thresholds, operational impact, zones, and capacity monitoring.
- Amazon Science: Dynamo, Amazon's Highly Available Key-value Store - primary paper for consistent hashing, object versioning, sloppy quorum, hints, and reconciliation.
- RFC 9562: Universally Unique IDentifiers - current UUID layouts, generation requirements, collision discussion, and security considerations.
- Google Cloud Spanner documentation: Schema design best practices - primary-key distribution, monotonic-key hotspots, UUID choices, and hash techniques.