Replication and Data Distribution

Introduction

Distributed databases store and process data across multiple machines, availability zones, regions, or countries. Distribution can improve throughput, fault tolerance, geographic responsiveness, and regulatory compliance, but it also introduces coordination costs. A distributed design must therefore answer four related questions:

1. Which copies of data exist, and where are they located?
2. When is a write considered committed?
3. Which node is allowed to accept a write?
4. How is data divided when one logical dataset becomes too large for a single server?

Replication creates multiple copies of the same logical data. Sharding, or partitioning, divides a dataset into subsets. They are frequently combined: a system may divide users across shards and replicate each shard across several regions.

The central engineering problem is not to identify a universally superior architecture. It is to select a design whose consistency, latency, availability, sovereignty, operational complexity, and cost match the workload and institutional obligations.

Learning objectives

After completing this chapter, the learner should be able to:

• distinguish synchronous and asynchronous replication;
• explain primary-replica, multi-primary, quorum, and leaderless models;
• evaluate geo-replication and sharding strategies;
• calculate basic quorum and replication relationships;
• analyse trade-offs among latency, availability, sovereignty, and cost;
• design an appropriate distribution strategy for a realistic workload;
• identify failure, consistency, governance, and operational risks.

1. Foundations of replication

A replica is a physically separate copy of data maintained by a database system. Replication may occur at the storage-block, log, row, document, or transaction level. Most modern systems replicate an ordered change stream, often called a write-ahead log, operation log, or commit log.

A replicated system has at least three distinct concepts:

• Data state: the values currently visible to clients.
• Replication state: which changes each replica has received and durably stored.
• Visibility state: which changes a replica is permitted to return to clients.

These states need not advance simultaneously. A replica may have received a change but not yet make it visible, or it may serve an older snapshot while catching up.

Synchronous and asynchronous replication

In synchronous replication, a write is acknowledged only after a specified set of replicas has confirmed the required operation, usually durable storage. Synchronous replication can provide strong protection against loss when a failure occurs, but the write path is constrained by the slowest required participant and by network distance.

In asynchronous replication, the primary acknowledges before all replicas confirm receipt. This usually reduces write latency and permits operation during some replica or network failures. However, a failure immediately after acknowledgement may cause acknowledged data to be absent from replicas that have not caught up.

The terms are not always absolute. Many systems implement hybrid policies: a transaction may require one local durable copy and one remote acknowledgement, while other replicas follow asynchronously.

For a synchronous acknowledgement set of replicas i = 1, …, k, an idealized lower bound on acknowledgement latency is:

where L_i is the round-trip network and processing latency to required replica i, and P represents durable processing and coordination overhead. The expression is a lower-bound model, not a performance guarantee. Queueing, retries, contention, elections, and serialization can make actual latency higher. In asynchronous replication, acknowledgement can often avoid the remote term, but replication lag becomes a risk variable.

Replication lag

Replication lag is the delay between a change becoming committed or visible at the source and becoming available at a target replica. It can be measured in time, log position, transaction sequence, or event count. A replica that is behind may return stale data, fail a read-after-write expectation, or require routing to a more current node.

A useful operational distinction is:

• Commit durability: whether a failure can erase an acknowledged write.
• Read freshness: how current the value returned to a client is.
• Transactional consistency: whether related operations preserve required ordering and atomicity.

These properties are related but not identical.

2. Primary-replica replication

In a primary-replica, or leader-follower, architecture, one node coordinates writes and replicas apply the resulting change stream. Clients generally send writes to the primary and may read from the primary or from replicas.

Figure: A primary distributes changes to synchronous and asynchronous replicas.

Strengths

Primary-replica designs are comparatively simple to reason about. A single ordering authority makes conflict prevention straightforward, and transactional applications can often retain familiar single-node semantics. Replicas can be used for read scaling, backups, reporting, and disaster recovery.

Limitations

The primary can become a write bottleneck and a failure-management focus. Automatic failover introduces the possibility of split brain, in which two nodes incorrectly believe they are primary. Replicas used for reads may be stale. Read scaling is also limited when transactions require a single authoritative view.

A safe failover protocol requires leader election, fencing or lease expiry, durable membership information, and a mechanism ensuring that an old primary cannot continue accepting writes. Merely detecting that a server is unreachable is insufficient: a network partition can make a healthy server appear failed.

3. Multi-primary replication

In a multi-primary system, two or more nodes accept writes. This can reduce geographic write latency and permit local operation during some inter-site failures. It also creates conflicts when different nodes modify the same logical record before learning about one another's changes.

Conflict handling may use:

• prevention through partition ownership or deterministic routing;
• last-write-wins, based on timestamps or logical clocks;
• field-level or operation-level merging;
• application-specific resolution;
• explicit rejection requiring human or service intervention.

Last-write-wins is simple but can silently discard valid updates, especially when clocks are inaccurate or an update with a later timestamp is semantically older. Conflict-free replicated data types can provide mathematically defined merges for selected data structures, but they do not solve arbitrary business conflicts.

Multi-primary replication is appropriate when writes are naturally partitionable, conflicts are rare or safely mergeable, and local write availability is more valuable than simple global ordering. It is risky for inventory, financial balances, or uniqueness constraints unless coordination is carefully designed.

4. Quorum replication

A quorum model represents a replicated object with N replicas. A write is accepted after W replicas acknowledge, and a read consults R replicas. A commonly used overlap condition is:

Under the assumptions that reads and writes contact the stated numbers of distinct replicas, that returned versions can be compared, and that at least one replica in the read set overlaps the acknowledged write set, the condition ensures overlap between a successful read and a successful write. It does not, by itself, guarantee linearizability, durable storage, conflict-free updates, or perfect freshness. Version selection, read repair, hinted handoff, and failure handling remain important.

For example, with N = 5, W = 3, and R = 3:

Thus the sets overlap under the stated model. A write quorum of 3 tolerates up to N − W = 2 replicas failing for write acknowledgement, while a read quorum of 3 tolerates up to N − R = 2 unavailable replicas for the read. These are availability bounds for a particular operation, not guarantees that the whole service remains available under every failure or partition.

A stronger requirement for a write quorum to survive f independent replica failures without losing the acknowledged value is commonly expressed as:

This assumes acknowledged copies are durable and failures are independent at the replica level. Correlated failures, such as a power or regional outage, invalidate a simple machine-count interpretation; failure domains must be represented explicitly.

Quorum replication is attractive because it offers tunable consistency and availability. However, contacting multiple replicas increases network traffic, tail latency, and operational complexity. It is especially important to distinguish a quorum of processes from a quorum of independent failure domains.

5. Leaderless replication

Leaderless systems allow clients or coordinators to send operations to multiple replicas without a permanently designated primary. Reads may query several replicas and reconcile versions; writes may be repaired when inconsistencies are detected.

Common mechanisms include:

• Read repair: a read identifies stale replicas and sends corrections.
• Hinted handoff: another node temporarily stores a change for an unavailable target.
• Anti-entropy repair: replicas periodically compare summaries or ranges and synchronize differences.
• Version vectors or logical clocks: metadata records causal or version relationships.

Leaderless architectures can be highly available and geographically flexible, but they shift complexity from a leader to conflict detection, metadata, repair, and client-visible anomalies. A successful write does not necessarily mean every future read immediately observes it. Applications must specify whether eventual consistency is acceptable.

6. Geo-replication and data sovereignty

Geo-replication places copies in different regions, countries, or legal jurisdictions. Its goals may include disaster recovery, lower user latency, business continuity, and compliance with data-residency requirements.

Latency and distance

A synchronous cross-region commit cannot be faster than the network round trip between the participating regions, plus processing time. Consequently, global synchronous writes can impose substantial latency. A common compromise is local synchronous replication within a region and asynchronous replication to a remote disaster-recovery region.

This design reduces normal write latency but introduces a recovery-point objective (RPO): the maximum acceptable amount of data that may be lost after a regional failure. If asynchronous replication is 20 seconds behind at the moment of failure, the practical RPO may approach 20 seconds, subject to measurement and workload conditions.

Recovery-time objective (RTO) is the maximum acceptable time to restore service. Low RTO generally requires pre-provisioned capacity, tested promotion procedures, replicated metadata, and well-defined DNS or routing changes. Replication alone does not guarantee rapid recovery.

Sovereignty and governance

Data sovereignty concerns the laws and governmental controls applicable to data because of its location, the customer, the processing activity, or the organization. A design must classify data and determine:

• where primary and backup copies may reside;
• whether encryption keys may cross borders;
• whether support personnel can access data remotely;
• how deletion, retention, and legal holds apply to every replica;
• whether telemetry and logs contain personal or regulated data.

A geographically local replica may improve latency but still violate policy if backups, snapshots, logs, or encryption keys are stored elsewhere. Sovereignty is therefore a complete data-lifecycle property, not merely a database-region setting.

7. Sharding and data distribution

Sharding divides a logical dataset into partitions called shards. Each shard may itself be replicated. The shard key determines placement and strongly influences scalability, locality, and load balance.

Shard-key strategies

Hash partitioning distributes keys relatively evenly and is effective for preventing hotspots. It makes range queries and co-location of related records more difficult.

Range partitioning assigns intervals, such as customer IDs or dates, to shards. It supports efficient range scans and archival but can create hotspots when new records concentrate in the latest range.

Geographic partitioning places records near users or legal jurisdictions. It can lower latency and support sovereignty, but users and transactions spanning regions become more difficult.

Directory-based partitioning uses a metadata service to map keys to shards. It permits flexible placement and migration but introduces a directory dependency and operational complexity.

A good shard key has high cardinality, even distribution, stable routing, and a relationship to dominant query patterns. It should not be chosen solely because it is a primary key.

Cross-shard operations

A transaction involving one shard is usually simpler and faster than a transaction spanning several shards. Cross-shard transactions may use two-phase commit, saga workflows, transactional messaging, or application-level compensation.

Two-phase commit has a prepare phase and a commit phase. Participants first promise that they can commit; after all are prepared, the coordinator announces the decision. It preserves atomicity under appropriate assumptions but adds latency and can leave participants blocked if the coordinator fails. Sagas decompose a business transaction into steps and define compensating actions; they improve flexibility but provide semantic rather than immediate atomicity.

Rebalancing

As data and traffic grow, shards must be split, merged, or moved. Safe rebalancing requires a consistent ownership map, incremental copying, change capture, validation, and controlled cutover. During migration, systems must avoid lost updates, duplicate processing, and inconsistent routing. Capacity planning should include temporary storage and network costs caused by movement.

8. Evaluating designs: latency, availability, sovereignty, and cost

A useful evaluation separates normal operation from failure operation.

A practical decision process is:

1. Classify data by criticality, sensitivity, residency, and consistency needs.
2. Identify read/write geography and peak traffic.
3. Define RPO and RTO for each failure domain.
4. Select the smallest coordination scope that satisfies correctness.
5. Choose shard keys from access patterns and growth projections.
6. Model failure, repair, migration, and regulatory deletion.
7. Test the design under latency, partition, overload, and replica loss.

A cost model should include more than database instance prices. If a dataset of size S is replicated with N full copies, raw storage is approximately N × S before indexes, logs, snapshots, compression differences, and temporary migration space. Network transfer may dominate when writes are replicated across regions. Operational cost includes monitoring, on-call response, consistency debugging, recovery exercises, and compliance evidence.

9. Consistency models and application contracts

Replication architecture should be selected with an explicit consistency contract. Important models include:

• Eventual consistency: if updates stop, replicas eventually converge, subject to system assumptions.
• Read-your-writes: a client can read its own successful updates.
• Monotonic reads: a client does not observe time moving backward across reads.
• Causal consistency: causally related operations are observed in causal order.
• Linearizability: each operation appears to take effect atomically between invocation and response, respecting real-time order.
• Serializability: concurrent transactions produce a result equivalent to some serial execution.

These properties address different questions. A system can provide serializable transactions within one shard while offering only eventual consistency across shards. Similarly, asynchronous replication may provide durable local commits while remote reads remain stale.

Applications should make consistency visible in APIs. For example, a payment service should not silently use a stale replica to decide whether funds are available. A product catalogue may tolerate stale descriptions but not stale inventory. Routing, session tokens, causal metadata, or bounded-staleness controls can help align reads with user expectations.

10. Applied case: a multinational retail platform

Background

Consider an illustrative retailer operating in Europe, Asia, and North America. Product browsing is read-heavy and tolerant of short-lived staleness. Orders, payments, and inventory reservations require stronger correctness. European customer records must remain within approved jurisdictions, and the business requires continued operation after a regional outage.

Problem

A single global primary produces high write latency for distant customers. A globally multi-primary design permits local writes but creates difficult conflicts for inventory and order identifiers. Read replicas alone do not solve the need for regional resilience or sovereignty.

Analysis

The workload should be divided by data semantics rather than by one universal policy:

• product descriptions and recommendations can use asynchronous geo-replication and cache-friendly eventual consistency;
• customer data can be sharded by approved residency region;
• inventory can be partitioned by fulfilment location, with a single authoritative owner for each stock pool;
• orders can be written in the customer's regional service, with immutable events replicated for analytics and controlled operational workflows.

Approach

Each regional cluster uses a primary-replica arrangement with synchronous replication across independent local failure domains. A remote region receives an asynchronous log stream for disaster recovery. The inventory shard for a warehouse has one write authority, preventing two regions from decrementing the same stock pool independently. Cross-domain business processes use a saga: reserve stock, authorize payment, create order, and compensate if a later step fails.

Reads requiring a customer's recent order are routed to the regional primary or to a replica known to have reached the relevant log position. Catalogue reads use asynchronous replicas. Encryption keys and backups follow residency policies, and deletion workflows propagate to replicas, indexes, caches, and analytical copies.

Outcome

The design accepts higher complexity in exchange for local user responsiveness, bounded correctness for inventory, regional recovery, and regulatory control. It does not provide instantaneous global consistency; that limitation is explicit and reflected in product and operational policies.

Learning points

1. Different data domains can legitimately use different replication models.
2. Ownership of a business invariant should be narrower than the entire global database when possible.
3. Disaster recovery replication must be tested, not merely configured.
4. Sovereignty applies to secondary copies and operational artifacts.
5. A consistency contract must be communicated to application developers and users.

11. Challenges, risks, and best practices

Common risks include stale reads, replication loops, log gaps, silent conflict loss, split brain, hotspot shards, unbounded metadata, and recovery procedures that work only in documentation. Monitoring should include replication lag, commit latency, quorum failures, conflict rates, shard skew, repair backlog, disk headroom, and cross-region transfer.

Best practices include:

• use independent failure domains rather than counting replicas on one host;
• document acknowledgement and read-routing semantics;
• fence old leaders during failover;
• test partitions and correlated failures;
• make writes idempotent where retries are possible;
• retain enough log history for lagging replicas to catch up;
• validate data after migration and repair;
• use immutable audit events for sensitive operations;
• rehearse RPO and RTO through controlled disaster exercises;
• treat deletion, retention, and key management as replicated workflows;
• avoid multi-primary writes for invariants that cannot be merged safely;
• choose shard keys using measured access patterns, not intuition alone.

Artificial intelligence may assist with hotspot prediction, replica placement, anomaly detection, and capacity forecasting. It should not autonomously weaken consistency or sovereignty controls. Any automated recommendation must be explainable, constrained by policy, and evaluated against failure scenarios.

Chapter Summary

Replication maintains multiple copies of data, while sharding divides data into independently managed partitions. Synchronous replication improves acknowledgement durability but adds coordination latency; asynchronous replication improves responsiveness and failure tolerance at the cost of lag and possible loss of remote copies. Primary-replica systems simplify ordering, whereas multi-primary and leaderless systems improve write locality but require conflict management.

Quorum designs use N replicas and read and write thresholds R and W. The condition R + W > N creates read-write overlap under explicit assumptions, but it is not by itself a guarantee of linearizability or durability. Geo-replication supports disaster recovery and local access, yet must be evaluated against network latency, RPO, RTO, transfer cost, and sovereignty obligations.

Sharding enables storage and throughput growth, but shard-key selection, cross-shard transactions, rebalancing, and hotspot control become central design concerns. The most reliable strategy is workload-specific: identify business invariants, define consistency and recovery contracts, select the smallest necessary coordination scope, and test the complete lifecycle—including failure, repair, migration, deletion, and compliance.

Key Terms

Replication: Maintaining multiple copies of logical data.

Replica: A copy of data maintained by a distributed system.

Synchronous replication: Replication in which required acknowledgements precede write success.

Asynchronous replication: Replication in which the source may acknowledge before all targets confirm receipt.

Replication lag: Delay between a source change and its availability at a target.

Primary: Node that coordinates writes in a primary-replica architecture.

Failover: Process of transferring service authority to another node.

Multi-primary: Architecture in which multiple nodes accept writes.

Conflict: Incompatible concurrent updates to the same logical data.

Quorum: Minimum set of replicas required for an operation to proceed.

Leaderless replication: Replication without a permanently designated write leader.

Read repair: Correcting stale replicas as part of a read operation.

Geo-replication: Replication across geographic regions or jurisdictions.

RPO: Maximum acceptable amount of data loss measured in time or recoverable state.

RTO: Maximum acceptable time to restore service.

Sharding: Dividing a dataset into independently managed partitions.

Shard key: Attribute or rule used to assign records to shards.

Hotspot: Disproportionate load concentrated on a shard or replica.

Data sovereignty: Legal and policy control associated with data location and handling.

Linearizability: Consistency property in which operations appear atomic and respect real-time order.

Review Questions

Short Answer Questions

1. What is the principal difference between synchronous and asynchronous replication?
2. Why can a primary-replica system return stale data to a client?
3. State the quorum overlap condition and explain its purpose.
4. What is replication lag, and why does it matter for disaster recovery?
5. Give two properties of a good shard key.

Long Answer Questions

1. Compare primary-replica, multi-primary, and leaderless replication in terms of ordering, conflict handling, availability, and operational complexity.
2. Explain how latency, RPO, RTO, and sovereignty influence geo-replication design.
3. Discuss the assumptions and limitations of the condition R + W > N.
4. Explain range, hash, geographic, and directory-based sharding, including appropriate use cases.
5. Describe a complete operational strategy for failover, repair, rebalancing, and compliance deletion in a replicated database.

Analytical / Application Questions

1. A five-replica object uses R = 3 and W = 3. Determine whether read-write overlap is guaranteed under the basic quorum model, and identify two guarantees that do not automatically follow.
2. Design a distribution strategy for a payment platform with global customers, strict transaction correctness, regional data-residency requirements, and a 30-minute disaster-recovery RTO. Justify replication mode, write authority, shard key, and recovery approach.
3. A retailer observes high latency from a globally synchronous multi-primary inventory system and frequent stock conflicts. Propose a revised architecture and explain how it changes latency, availability, consistency, sovereignty, and cost.

Multiple Choice Questions with Answers

Question 1

Which statement best describes asynchronous replication?

A. A write is acknowledged only after every replica has committed it

B. A source may acknowledge a write before all replicas confirm receipt

C. Replicas never store a write log

D. All replicas must be in the same geographic region

Correct Answer: B. A source may acknowledge a write before all replicas confirm receipt

Question 2

In a primary-replica architecture, which node normally coordinates writes?

A. A cache node

B. A read-only replica

C. The primary

D. The backup archive

Correct Answer: C. The primary

Question 3

For N = 5, R = 3, and W = 3, which quorum relationship holds?

A. R + W < N

B. R + W = N

C. R + W > N

D. R × W = N

Correct Answer: C. R + W > N

Question 4

What is a principal risk of multi-primary replication?

A. It cannot support local writes

B. Concurrent writes may create conflicts

C. It always requires synchronous global commits

D. It prevents geographic distribution

Correct Answer: B. Concurrent writes may create conflicts

Question 5

Which sharding strategy generally supports efficient range scans?

A. Range partitioning

B. Pure random replication

C. Leader election

D. Read repair

Correct Answer: A. Range partitioning

Question 6

What does RPO measure?

A. Maximum acceptable service-restoration time

B. Maximum acceptable data loss after a failure

C. Number of replicas required for a read

D. Average query-processing time

Correct Answer: B. Maximum acceptable data loss after a failure

Question 7

Which statement about R + W > N is correct?

A. It alone guarantees linearizability

B. It guarantees that every replica is current

C. It provides read-write set overlap under stated assumptions

D. It eliminates the need for conflict resolution

Correct Answer: C. It provides read-write set overlap under stated assumptions

Question 8

Which design most directly prevents two regions from independently decrementing the same inventory pool?

A. Assign one authoritative write owner to that inventory shard

B. Use more catalogue caches

C. Use last-write-wins without versioning

D. Remove all replicas

Correct Answer: A. Assign one authoritative write owner to that inventory shard

Question 9

Why can cross-region synchronous replication increase write latency?

A. It requires acknowledgement from a geographically distant participant

B. It eliminates all network communication

C. It converts every write into a read

D. It reduces the number of required acknowledgements to zero

Correct Answer: A. It requires acknowledgement from a geographically distant participant

Question 10

Which item is part of data-sovereignty analysis?

A. Only the location of the primary database server

B. Locations of replicas, backups, logs, and encryption keys

C. Only the colour of the monitoring dashboard

D. Only the number of application users

Correct Answer: B. Locations of replicas, backups, logs, and encryption keys