diff --git a/manifest.json b/manifest.json index a367ec5..8e334d7 100644 --- a/manifest.json +++ b/manifest.json @@ -133,6 +133,10 @@ "title": "Distributed-Systems Fundamentals, Time, Ordering, and Consistency", "path": "systemdesign/distributed-systems-time-ordering-consistency" }, + { + "title": "Replication, Partitioning, Sharding, and Quorum Systems", + "path": "systemdesign/replication-partitioning-sharding-quorum-systems" + }, { "title": "Coordination, Consensus, Leader Election, Leases, and Distributed Locks", "path": "systemdesign/coordination-consensus-leases-locks" diff --git a/src/content/systemdesign/replication-partitioning-sharding-quorum-systems.html b/src/content/systemdesign/replication-partitioning-sharding-quorum-systems.html new file mode 100644 index 0000000..8ee271d --- /dev/null +++ b/src/content/systemdesign/replication-partitioning-sharding-quorum-systems.html @@ -0,0 +1,1715 @@ +

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. +

+ +
+ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
RequirementLikely techniqueNew cost or risk
Survive a machine or zone lossReplicas across independent failure domainsWrite coordination, lag, failover, extra storage
Scale read throughputRead replicas or independently readable replicasStaleness, consistency routing, replica saturation
Exceed one node's storage or write capacityHorizontal shardingRouting, cross-shard operations, rebalancing
Place data near users or satisfy residency rulesRegion-aware replication and placementWide-area latency, partitions, policy complexity
Keep independent data lifecycles or access controlsVertical partitioningJoins and atomic updates may cross boundaries
+
+ +

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

+ + +

+ 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. +
  3. + The leader appends the change to its log. Durability depends on the configured flush point. +
  4. +
  5. Followers fetch or receive log records, persist them, and apply them in order.
  6. +
  7. The leader acknowledges after its configured local and remote conditions are satisfied.
  8. +
  9. Followers expose applied state to reads only under their serving policy.
  10. +
+ +

+ 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

+
+ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
ModeSuccess usually waits forBenefitMain trade-off
AsynchronousLeader persistence onlyLower latency and remote failure isolationA failover can lose acknowledged but unreplicated writes
Synchronous receiveRemote receipt or log persistenceBetter failure toleranceDoes not always mean remote apply or read visibility
Synchronous applyRemote apply to queryable stateStronger immediate read behaviorWrite latency includes slow follower apply
Geographic synchronousAnother region or fault domainLow data-loss objective across regional lossWide-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: +

+ + +

+ 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

+ + +

+ 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 policyAvailabilityLatencySuitable example
ONEHigh while any suitable replica respondsUsually lowApproximate feed counters where staleness is acceptable
QUORUMTolerates a minority of failuresWaits for a majority and reconciliationUser-visible profile state with bounded conflict handling
ALLAny unavailable replica can blockBound by slowest replicaRare operations where every replica must observe before success
LOCAL_QUORUMSurvives local minority failureAvoids wide-area pathRegion-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

+ + +

+ 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:

+ + +
+
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. +

+ +
+ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
SplitGood fitWeakness
TenantIsolation, residency, tenant-local queriesOne large tenant can dominate a shard
Entity ID hashEven point reads and writesRange queries scatter
Time rangeRetention, recent-window queries, archivalCurrent range becomes a write hotspot
Feature or columnsSecurity and independent scalingCross-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

+ + +
+ 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:

+ + +

Corrections by hotspot type

+
+ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
CauseSafer correctionTrade-off
Monotonic leading keyHash prefix, time buckets with sub-buckets, or randomized IDRange reads must merge buckets
Single celebrity or room keyRead cache, replicated materialization, or split derived dataInvalidation and merge complexity
Large tenantDedicated placement or sub-shard within tenantSpecial routing and operational policy
Expensive query classAdmission control, precomputation, or workload isolationFreshness or feature limits
Unequal node capacityWeighted placement and capacity-aware balancingMore 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

+ + +

+ 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. +

+ +
+ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
StrategyStrengthFailure or costGood fit
Single database sequenceSimple uniqueness and total allocation orderCentral dependency, contention, visible volumeOne write authority with moderate allocation rate
Range or hi-lo allocationClients allocate locally from reserved blocksGaps, stranded ranges, allocation-service recoveryShards needing numeric IDs without per-ID coordination
UUIDv4Decentralized, random 122-bit payloadLarge index, random locality, collision still probabilisticPublic opaque IDs and independent writers
UUIDv7Standard time-ordered layout with random bitsLeaks creation time and needs rollback handlingRough time locality when timestamp exposure is acceptable
Snowflake-style integerCompact, decentralized, roughly time orderedWorker-ID coordination, clock rollback, topology leakageHigh-rate internal event or entity creation
Random tokenUnpredictability with sufficient entropyLonger representation and collision calculationCapability 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

+ + +

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

+ + +
+
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

+ + +

Observability

+
+ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
AreaSignalsDiagnostic question
Replicationlag bytes/time, apply rate, log retention, under-replicated groupsIs data received but not applied, or not received?
Quorumresponses by consistency level, unavailable, timeout, conflict rateWhich replica or domain prevents enough evidence?
Distributionbytes, QPS, CPU, disk, p99 latency by partition and tenantIs imbalance caused by data, traffic, or request cost?
Repairhint age, replay rate, repair coverage, mismatched rangesCan replicas converge before tombstones expire?
Migrationbytes copied, catch-up lag, checksum failures, stale-map redirectsIs cutover safe and have routers converged?
IDsworker lease conflicts, clock rollback, sequence exhaustion, collisionsCan two generators issue the same bit tuple?
+
+ +

Troubleshooting playbook

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

18. Security, privacy, abuse, and trust boundaries

+

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

+ + + +

+ 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

+
+ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
ScenarioUnsafe reactionSafer response
Leader unreachable from one zonePromote a leader on every sideAllow only an authorized quorum side and fence the old epoch
Follower lag growsKeep sending freshness-sensitive readsRemove from that read class, diagnose receive versus apply lag
Quorum write times outRetry with a new operation IDRetry idempotently or query outcome by original ID
Returning replica has old valuesTrust its process healthKeep non-serving until catch-up and repair prove consistency
Hot partitionAdd replicas expecting writes to spreadSplit write ownership or change key design if semantics allow
Stale partition mapAccept write at old ownerReject stale epoch, refresh, and retry idempotently
Migration interruptedDelete source because copy startedResume durable phase state; source remains authority until cutover
Clock moves backward on ID workerContinue with reset sequenceStop, wait safely, or use a proved logical-time fallback
Repair after tombstone expiryMerge old value as liveKeep repair interval below retention and replace stale nodes safely
+
+ +

21. Common mistakes and corrections

+ + +

22. Testing strategy

+ +

Unit and property tests

+ + +

Integration and compatibility tests

+ + +

Fault, partition, and recovery tests

+ + +

Load and capacity tests

+ + +

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

+ + +

26. Current primary official references

+ + +