Distributed Computing

Chapter 6
Replication and Consistency

Post Graduate Program in IT Engineering
Distributed Computing   •   Post Graduate

Replication and Consistency

Introduction

Replication is the maintenance of multiple copies of data or service state on different computing nodes. It is a central technique in distributed computing because a single machine is vulnerable to hardware failure, software defects, network disconnection, overload, and geographic disasters. Replication can improve availability, read performance, fault tolerance, and recovery time. It also introduces a fundamental difficulty: copies may temporarily or permanently disagree.

Consistency concerns the relationship between operations observed by clients and the states maintained by replicas. Stronger consistency usually requires more coordination, which can increase latency and reduce availability during network partitions. Weaker consistency can improve responsiveness and scalability, but it transfers complexity to application developers, conflict-resolution mechanisms, and user-interface design.

This chapter examines primary–backup replication, active replication, quorum systems, anti-entropy, consistency models, conflict resolution, CRDTs, and session guarantees. The objective is not to identify one universally superior model, but to develop the ability to select and combine mechanisms according to failure assumptions, workload, correctness requirements, and business risk.

Learning Objectives

After completing this chapter, a student should be able to:

1. Explain why replication is used and identify the consistency problems it creates.

2. Design and compare primary–backup and active-replication architectures.

3. Calculate quorum conditions and reason about read–write and write–write overlap.

4. Explain anti-entropy, hinted handoff, and read repair.

5. Distinguish linearizability, sequential consistency, causal consistency, and eventual consistency.

6. Select conflict-resolution methods and assess their limitations.

7. Explain the operation of CRDTs and session guarantees.

8. Evaluate replication designs in terms of availability, performance, fault tolerance, and governance.

1. Replication Fundamentals

A replica is a process or storage system that maintains a copy of an object, partition, or service state. Replication may be:

  • Synchronous, where an operation is acknowledged only after specified replicas have recorded it.
  • Asynchronous, where the client may receive an acknowledgement before all replicas receive the update.
  • Full, where each replica stores the complete dataset.
  • Partial or partitioned, where each node stores only selected shards, possibly with several replicas per shard.
  • State-machine replication, where replicas execute the same ordered commands.
  • Data replication, where replicas exchange database records, log entries, or object updates.

Replication provides several benefits. A service can continue after a node failure, place copies near users, distribute read traffic, and recover from corruption. It can also support maintenance without complete downtime. However, the system must answer difficult questions: Which copy accepts a write? When is a write durable? What happens if replicas receive updates in different orders? How are conflicting updates merged? Which failures are tolerated?

A useful design distinction is between safety and liveness. Safety means that the system does not produce an unacceptable result, such as acknowledging two incompatible committed values. Liveness means that legitimate operations eventually complete. During a network partition, some systems preserve strong safety by rejecting operations; others preserve availability by accepting operations and reconciling later. This tension is related to the CAP perspective: under a partition, a system cannot simultaneously guarantee both availability and strong consistency for all operations.

Replication also has a failure model. Crash-stop systems assume that a failed process stops permanently; crash-recovery systems allow it to restart; Byzantine systems allow arbitrary or malicious behavior. Most practical database replication uses crash and network-failure assumptions, while safety-critical or adversarial environments may require Byzantine fault tolerance.

2. Primary–Backup Replication

Architecture and operation

In primary–backup replication, one replica is designated the primary (or leader), and one or more replicas are backups (or followers). The primary coordinates writes and usually establishes their order. A typical write path is:

1. A client sends a request to the primary.

2. The primary validates the request and assigns it a sequence number.

3. The update is transmitted to backups, often through a replication log.

4. The primary waits for the required acknowledgement condition.

5. The primary replies to the client.

6. Backups apply the ordered update and report progress.

The acknowledgement condition determines durability and availability. In a synchronous design, the primary may wait for every backup or for a quorum of backups. In an asynchronous design, it may reply after local persistence, accepting a risk that an unreplicated update will be lost if the primary fails.

Reads can be directed to the primary, which simplifies consistency, or to backups. Backup reads require a freshness policy. A client might tolerate stale data, require a replica to have applied a specified log position, or use leases and timestamps to constrain staleness.

Failover and split-brain prevention

When the primary fails, a coordinator or consensus protocol selects a new primary. A safe failover mechanism must ensure that an old primary cannot continue accepting writes after losing authority. Otherwise, a network partition can create two active primaries, a condition called split brain.

Systems commonly use epochs, terms, fencing tokens, leases, or consensus protocols such as Raft or Paxos to establish authority. A fencing token is a monotonically increasing identifier presented to a storage system; operations from an old primary with a lower token are rejected. Time-based leases require carefully bounded clock uncertainty and are unsafe if a paused process can resume after its lease has expired without a fencing mechanism.

Strengths and limitations

Primary–backup replication offers a simple write-ordering model and is often easy to reason about. It is appropriate for systems requiring a single serialization point, such as transactional databases and metadata services. Its weaknesses include primary bottlenecks, failover complexity, and possible unavailability when the primary cannot contact the required backups. Asynchronous primary–backup replication may lose recent acknowledged operations unless the system uses durable logs, quorum acknowledgement, or a recovery process that preserves the primary's log.

3. Active Replication and State-Machine Replication

In active replication, every replica executes each request rather than receiving a completed state update from a primary. A communication layer delivers commands in the same order to all non-faulty replicas. If the service is deterministic and starts from equivalent state, replicas produce equivalent results.

The essential requirements are:

  • Deterministic execution: the same ordered input produces the same output and state transition.
  • Agreement on order: correct replicas deliver requests in a common total order, or in an order sufficient for the application.
  • Deterministic external effects: interactions with clocks, random numbers, files, and external services must be controlled or replicated.

For example, a service must not independently call a local random-number generator while processing a replicated command. Instead, a random value can be generated as part of the ordered command. Similarly, a timestamp should be supplied by the ordering layer or treated as replicated input.

Active replication can tolerate a replica failure without replaying a large state-transfer log, and it distributes request execution. It may, however, consume substantial CPU and network capacity because every replica executes every command. Nondeterminism, floating-point differences, concurrency, and external side effects make implementation challenging. Byzantine state-machine replication adds cryptographic authentication and larger quorum requirements because replicas may behave maliciously.

Primary–backup and active replication are not absolute opposites. A primary can order commands while all replicas execute them, combining leader-based ordering with state-machine replication. Conversely, a primary can send state changes rather than commands. The important design question is whether replicas independently execute an agreed operation or passively apply a replicated result.

4. Quorum Replication

A quorum system stores an item on a replication group of size N. A write must be accepted by at least W replicas, and a read must consult at least R replicas. If

R + W > N,

then every qualifying read set and every qualifying write set overlap in at least one replica. If

2W > N,

then two successful write sets overlap, which is useful for preventing concurrent writes from being treated as completely independent under a single-version protocol. These conditions assume that replicas return sufficiently current versions, that version comparison is meaningful, and that the protocol resolves concurrent values correctly. Quorum overlap alone does not automatically provide linearizability.

For example, with N = 5, R = 3, and W = 3, read and write sets overlap because 3 + 3 > 5. Two writes also overlap because 2(3) > 5. If W = 2 and R = 2, neither condition is guaranteed: two writes can use disjoint sets, and a read can miss a recent write. Such a configuration may still be useful for highly available eventual-consistency workloads, but it does not by itself guarantee strong read semantics.

A quorum protocol must define version selection, repair, and failure behavior. A read may return the newest version according to a logical timestamp, vector clock, or application-specific ordering. If different versions are concurrent, returning one arbitrarily can lose information. Some systems return all conflicting versions to an application resolver. A write may be accepted by W replicas and later propagated to others through anti-entropy.

Quorums can be weighted, allowing replicas with different capacities or failure domains to contribute different weights. Quorum design should also consider correlated failures. Five replicas in one rack do not provide the same resilience as five replicas distributed across independent availability zones.

5. Anti-Entropy and Replica Repair

Anti-entropy is a background process through which replicas compare their state and exchange missing updates. It is especially important in eventually consistent systems, where foreground operations may complete without contacting every replica.

A naive anti-entropy protocol compares every record and is expensive. Practical systems use Merkle trees, version vectors, logs, checksums, or partition summaries. A Merkle tree allows two replicas to compare hashes from the root downward and identify differing ranges without transmitting all identical data. Version vectors summarize which updates a replica has observed from different writers; they can help identify missing or concurrent updates, although metadata may grow with the number of writers.

Related mechanisms include:

  • Read repair: while serving a read, the system updates stale replicas discovered during the read.
  • Hinted handoff: a temporary surrogate stores a write for an unavailable target replica and forwards it later.
  • Gossip: nodes periodically exchange membership, failure, or state information with selected peers.
  • Log shipping: a replica receives an ordered stream of changes from another node.

Anti-entropy improves convergence but does not by itself guarantee a bounded convergence time. Delays, permanent partitions, repeated failures, overloaded repair processes, and unresolved conflicts can prevent convergence. Repair should be monitored using metrics such as replication lag, anti-entropy backlog, version divergence, and time to convergence.

6. Consistency Models

A consistency model specifies what values and operation order a client may observe. It is a contract between the distributed system and its applications.

Linearizability

Linearizability requires that each operation appear to take effect atomically at some point between its invocation and response, and that the resulting single order respects real-time precedence. If operation A completes before operation B begins, A must precede B in the linearized history.

Linearizability is intuitive for individual objects such as registers, locks, and leader-election records. It supports reasoning as if there were one current copy. It often requires coordination and can be unavailable during a partition. Linearizability of individual objects does not automatically provide serializable multi-object transactions.

Sequential consistency

Sequential consistency requires that the result be equivalent to some sequential execution that preserves the program order of each individual process. Unlike linearizability, it need not preserve real-time order between operations issued by different processes. Thus, it is weaker than linearizability but stronger than many asynchronous models.

Causal consistency

Causal consistency preserves the order of causally related operations. If a client reads a message and then posts a reply, a system offering causal consistency ensures that another observer does not see the reply without being able to see the message on which it depends. Concurrent operations need not be observed in the same order everywhere.

Causality can arise from program order, reads-from relationships, and transitive chains. Implementations may track dependencies using vector clocks, dotted version vectors, dependency metadata, or session context. Causal consistency is valuable for collaborative systems, social feeds, and replicated workflows where preserving “because of” relationships matters more than establishing one global order.

Eventual consistency

Eventual consistency generally means that, if updates stop and communication and repair continue, replicas eventually converge to the same value. It does not specify which value wins, how quickly convergence occurs, or what a client sees before convergence. A system can be eventually consistent yet offer additional guarantees such as monotonic reads or read-your-writes.

Eventual consistency is attractive for geo-distributed services and intermittently connected applications. It requires explicit conflict policies and careful user-experience design. “Last write wins” is simple but can silently discard a legitimate concurrent update, especially when clocks are inaccurate or timestamps have coarse resolution.

Comparative view

Linearizability provides the strongest single-object real-time semantics but may impose the greatest coordination cost. Sequential consistency preserves each client's order but not necessarily global real time. Causal consistency preserves meaningful dependency order while allowing concurrent operations to proceed independently. Eventual consistency emphasizes convergence and availability but leaves more responsibility to reconciliation and applications. These models are not merely performance settings; they determine what application-level reasoning is valid.

7. Conflict Resolution and CRDTs

A conflict occurs when concurrent or independently accepted updates cannot be applied in one unambiguous order. Resolution strategies include:

  • Last-write-wins: select a version using a timestamp or logical ordering. This is easy to implement but may lose updates.
  • Application merge: preserve both values or apply domain-specific rules, such as combining shopping-cart items.
  • Multi-value registers: retain concurrent values and require a later resolver.
  • Operational transformation: transform concurrent operations so that text edits remain meaningful; this requires complex transformation rules.
  • CRDTs: design state or operations so that replicas merge without coordination and converge under suitable delivery assumptions.

A state-based CRDT provides a join operation on states. States form a join-semilattice: the merge operation is associative, commutative, and idempotent. These properties mean that messages may be reordered, duplicated, or delivered more than once without changing the final result, provided updates move state monotonically with respect to the CRDT's ordering.

An operation-based CRDT disseminates operations. Correctness generally requires reliable eventual delivery and, depending on the data type, assumptions about causal delivery or operation metadata. Operation-based designs can use less bandwidth than repeatedly sending full state, but delivery and duplicate suppression become important.

Examples include a grow-only set, an observed-remove set, a grow-only counter, and a positive-negative counter. A grow-only counter can be represented by a vector of per-replica counts; each replica increments its own component, and merging takes the component-wise maximum. The total is the sum of the components. This avoids double-counting repeated state dissemination. Decrements require additional structure, such as separate increment and decrement components.

CRDTs do not solve every conflict. They may consume substantial metadata, cannot automatically encode business invariants such as “account balance must not become negative,” and may require tombstone retention for removals. Coordination or escrow techniques are often necessary for non-monotonic constraints.

8. Session Guarantees

System-wide consistency can be weaker while still providing useful guarantees within a client session. Common session guarantees are:

  • Read-your-writes: after a client successfully updates an object, later reads in that session do not return an older version.
  • Monotonic reads: once a session has observed a version, later reads do not move backward to an older version.
  • Monotonic writes: writes from one session are applied in the order issued by that session.
  • Writes-follow-reads: a write based on information observed in a read is not applied before the version it depends on.

These guarantees can be implemented with session tokens, version vectors, replica routing, or minimum required log positions. For example, after a write, the server can return a token representing the write's log position. A subsequent read presents the token, and the chosen replica delays or forwards the read until it has caught up. Session guarantees are particularly important in user-facing applications: a user should not see a newly uploaded photograph disappear merely because a later request reached a lagging replica.

9. Worked Design Example: A Geo-Distributed Inventory Service

Consider an online retailer with customers in three regions. Product descriptions and reviews can tolerate seconds of staleness, but inventory reservation cannot permit the same unit to be sold twice.

A suitable design separates data according to invariants rather than applying one consistency model to the entire application.

Catalog and reviews: These can use asynchronous replication, quorum reads where useful, anti-entropy, and application-level merging. Review additions can be represented as an append-oriented data type or an add-wins set. Temporary divergence is acceptable if users eventually see the same content.

Inventory reservation: A reservation requires a coordination protocol. One option is a leader-based partition for each stock-keeping unit, with synchronous replication to a quorum and a fencing token during failover. Another is escrow: each region receives a bounded allocation of units and may sell only from its allocation. Escrow improves local availability but requires rebalancing when demand changes. A CRDT counter alone is unsafe for this invariant because independent decrements can exceed available stock.

Customer session: After placing an order, the customer receives a session token. Order-history reads use read-your-writes and monotonic reads so that the confirmation is not followed by an older view. Catalog reads may be served from nearby replicas without the same guarantee.

Failure analysis: If a regional network partition occurs, catalog browsing continues. Inventory reservations may be rejected or limited to local escrow, preserving the no-overselling invariant. If a replica is delayed, anti-entropy repairs it after connectivity returns. If a primary fails, a consensus-backed election and fencing token prevent the old primary from issuing reservations.

The key outcome is not a single global consistency level. It is a deliberately mixed architecture in which strong coordination protects the business invariant and weaker mechanisms provide scalable availability for less critical data.

10. Engineering, Governance, and Contemporary Issues

Replication design should begin with explicit invariants and failure assumptions. Teams should document whether acknowledged writes may be lost, the maximum tolerated staleness, the required behavior during partitions, and the recovery procedure. Testing should include delayed messages, duplicated messages, reordered messages, node pauses, clock skew, partial disk failure, and repeated failover.

Operational observability is essential. Useful indicators include p95 and p99 replication lag, quorum failure rate, stale-read frequency, conflict rate, repair backlog, failover duration, and divergence between replicas. Disaster-recovery exercises should measure the actual recovery point objective and recovery time objective rather than relying on configuration claims.

Replication may multiply the locations where personal or regulated data is stored. Data residency, encryption, key management, retention, deletion, auditability, and access-control requirements must therefore be applied to every replica and backup. A deletion request is not complete merely because the primary copy was deleted; repair logs, snapshots, caches, and offline backups require a defined lifecycle.

AI-assisted applications create additional consistency concerns. A recommendation or generated response based on stale user data may be misleading, while an automated agent might repeat an operation after a timeout. APIs should use idempotency keys, version checks, and explicit freshness metadata. Generated code must not be trusted to infer consistency guarantees from a product name; those guarantees must be verified from protocol documentation and tests.

Current developments include geo-replicated databases with tunable consistency, hybrid logical clocks, causal databases, serverless stateful services, edge replication, and formally verified replication protocols. These developments improve flexibility but do not eliminate fundamental trade-offs. More sophisticated metadata and coordination often shift cost from latency to storage, operations, or application complexity.

Best-Practice Design Principles

1. Define business invariants before selecting a consistency model.

2. Separate data with different correctness and availability requirements.

3. Make failover authority explicit and prevent split brain with fencing or consensus.

4. Treat quorum overlap as a necessary property in some designs, not as proof of linearizability.

5. Use idempotent operations and unique request identifiers to survive retries.

6. Use logical versions or causal metadata when wall-clock timestamps are unsafe.

7. Make conflicts visible when silent loss would be harmful.

8. Bound and monitor replica lag and repair backlog.

9. Test partitions, pauses, reordering, duplication, and recovery—not only normal operation.

10. Include replicas, logs, caches, and backups in security, privacy, and deletion policies.

Chapter Summary

Replication improves fault tolerance, availability, locality, and performance, but creates the problem of coordinating and reconciling multiple copies. Primary–backup replication centralizes ordering and simplifies reasoning, while active or state-machine replication has every replica execute an agreed sequence of deterministic commands. Quorum systems trade read and write coordination against availability; the conditions R + W > N and 2W > N describe overlap under stated assumptions but do not alone guarantee linearizability. Anti-entropy, Merkle trees, read repair, hinted handoff, and gossip support convergence and recovery.

Linearizability preserves real-time operation order, sequential consistency preserves each client's program order, causal consistency preserves dependencies, and eventual consistency focuses on convergence after updates cease. Conflict resolution may use timestamps, application merges, multi-value records, operational transformation, or CRDTs. CRDTs exploit mathematically well-behaved merge operations but cannot automatically preserve every business invariant. Session guarantees provide valuable user-level behavior even when global consistency is weak. Effective designs combine mechanisms selectively, protect critical invariants with coordination, and treat observability, privacy, security, and recovery as parts of replication engineering.

Key Terms

Replica: A process or storage copy maintaining duplicated service or data state.

Primary–backup replication: A design in which a primary orders or accepts updates and backups maintain copies.

Active replication: Replication in which replicas execute the same ordered requests.

State-machine replication: Deterministic execution of an agreed command sequence by replicas.

Quorum: A sufficient subset of replicas required to accept an operation.

Linearizability: A consistency model preserving real-time order as if operations were atomic.

Sequential consistency: A model preserving each process's operation order in some global sequential history.

Causal consistency: A model preserving the order of causally related operations.

Eventual consistency: Convergence of replicas if updates stop and communication continues.

Anti-entropy: Background comparison and exchange of replica state.

Read repair: Repair of stale replicas discovered during a read.

Hinted handoff: Temporary storage of an update on behalf of an unavailable replica.

Conflict: Incompatible or concurrent versions requiring reconciliation.

CRDT: A conflict-free replicated data type designed for convergent distributed updates.

Session guarantee: A consistency property maintained for operations within a client session.

Fencing token: An authority value used to reject operations from an obsolete primary.

Split brain: A failure condition in which multiple nodes incorrectly act as primary.

Replication lag: The delay between an update's acceptance and its application at a replica.

Review Questions

Short Answer Questions

1. Why does replication improve availability, and why does it create consistency problems?

2. What is the purpose of a fencing token in primary–backup replication?

3. State the quorum conditions R + W > N and 2W > N and explain their significance.

4. Distinguish causal consistency from eventual consistency.

5. What is read-your-writes session consistency?

Long Answer Questions

1. Compare primary–backup replication and active state-machine replication, including their assumptions, advantages, and limitations.

2. Explain anti-entropy, Merkle trees, read repair, hinted handoff, and gossip as mechanisms for convergence.

3. Compare linearizability, sequential consistency, causal consistency, and eventual consistency using suitable application examples.

4. Explain state-based and operation-based CRDTs, including their algebraic properties and limitations.

5. Develop a replication strategy for a geo-distributed transactional service with both critical and non-critical data.

Analytical / Application Questions

1. A five-replica system uses R = 2 and W = 3. Determine whether read/write overlap and write/write overlap are guaranteed. Explain what additional assumptions are needed before drawing consistency conclusions.

2. Design a reservation service that must never oversell inventory during a regional partition. Compare quorum-leader replication with escrow allocation and justify your choice.

3. An application displays a newly submitted post and then, after refresh, shows an older page that omits it. Identify the violated session guarantee and propose two implementation remedies.

Multiple Choice Questions with Answers

Question 1

In a quorum system with N = 5 replicas, which configuration guarantees both read/write overlap and write/write overlap under the standard quorum assumptions?

A. R = 2, W = 2

B. R = 3, W = 2

C. R = 3, W = 3

D. R = 1, W = 3

Correct Answer: C. R = 3, W = 3

Question 2

Which consistency model requires operations to appear atomic while preserving real-time precedence?

A. Eventual consistency

B. Causal consistency

C. Sequential consistency

D. Linearizability

Correct Answer: D. Linearizability

Question 3

What is the primary purpose of anti-entropy?

A. To elect a primary using clock synchronization

B. To compare replicas and exchange missing updates

C. To prevent all concurrent writes

D. To encrypt replicated data

Correct Answer: B. To compare replicas and exchange missing updates

Question 4

Which condition is essential for deterministic state-machine replication?

A. Every replica uses an independent random number generator

B. Replicas execute requests in a common order and deterministically

C. Only backups execute client requests

D. Replicas may apply commands in arbitrary order

Correct Answer: B. Replicas execute requests in a common order and deterministically

Question 5

A client writes a value and then reads an older value in the same session. Which guarantee has been violated?

A. Read-your-writes

B. Monotonic writes

C. Writes-follow-reads

D. Eventual convergence

Correct Answer: A. Read-your-writes

Question 6

What property of a state-based CRDT makes repeated or reordered state dissemination safe?

A. The merge operation is associative, commutative, and idempotent

B. All updates use synchronized wall-clock timestamps

C. Every update requires a global lock

D. Only one replica is allowed to modify state

Correct Answer: A. The merge operation is associative, commutative, and idempotent

Question 7

Why is quorum overlap alone insufficient to guarantee linearizability?

A. Quorums cannot contain more than one replica

B. Overlap does not define freshness, ordering, or conflict handling

C. Linearizability applies only to backups

D. Quorums work only during network partitions

Correct Answer: B. Overlap does not define freshness, ordering, or conflict handling

Question 8

Which mechanism most directly prevents an obsolete primary from continuing to issue writes after failover?

A. A cache expiration time alone

B. A larger read quorum only

C. A fencing token enforced by the storage system

D. A client-side retry loop

Correct Answer: C. A fencing token enforced by the storage system

Question 9

With N = 5, R = 2, and W = 3, is read/write overlap guaranteed?

A. Yes, because R + W = 5

B. Yes, because 2W > N

C. No, because R + W = 5 is not greater than N

D. No, because W must always equal N

Correct Answer: C. No, because R + W = 5 is not greater than N

Question 10

Why might a CRDT be unsuitable by itself for preventing inventory overselling?

A. CRDTs cannot replicate across regions

B. CRDTs always require a single primary

C. Independent decrements may violate a non-negative stock invariant

D. CRDTs cannot merge concurrent updates

Correct Answer: C. Independent decrements may violate a non-negative stock invariant

Distributed Computing
Chapter 6 of 15