06-Replicated State Machine & Primary Backup

Replicated State Machine Background

A service running on one machine has an obvious weakness:

One machine fails → service stops

The natural solution is to replicate service data:

        ┌─ Replica 1
Service ├─ Replica 2
        └─ Replica 3

But merely copying data is insufficient.

If the replicas:

• contain different state;
• process requests in different orders;
• disagree about which requests completed;
• produce different outputs after failover;

then switching to a backup can expose incorrect behavior.

The real goal is:

Replicate the service so that all replicas behave like one logical machine, even when an individual machine fails.

This is the purpose of a replicated state machine, or RSM.

Fault Tolerance and Availability

Fault tolerance means the service continues behaving correctly despite failures.
The desired user experience is often failure-agnostic:

• the service remains available;
• completed operations are not lost;
• the user does not see state move backward;
• failover does not produce contradictory results;
• the performance disruption is limited.

Suppose a client successfully received OK for the following request:

deposit(account, $100)

If the primary then crashes, the new primary must not behave as though the deposit never occurred. The previously visible result is now part of the service’s externally observed history.

Failure Models

A failure model specifies what faulty components are allowed to do.

Fail-stop failure

When a node stops, every other node knows with certainty that it failed

This is the simplest model.
In reality, perfectly detecting failure is difficult because failure may cause due to a lot of things:

• the node actually failed
• merely slow
• network partition

Crash failure

When a node stops, other nodes may not know whether it crashed

This creates uncertainty:

Crashed vs slow vs network partitioned

This is the uncertainty behind FLP and many leader-election problem

Omission failure

A node may fail to send or receive some messages

For example:

• it processes a client request but loses the response;
• it sends an update to one backup but not another;
• it receives only part of the replication stream.

Byzantine failure

A faulty node can behave arbitrarily

It may:

• send different messages to different replicas;
• invent operations;
• corrupt its state;
• lie about what it committed;
• impersonate a correct execution;
• collude with other faulty nodes.

How woud you tolerate a failed node?

• Replicas
  • I have some backups of my data, so will not lose data
• Failure-agnostic
  • Ideally, do not want users to realize there was a failure at all

How would you make failures agnostic?

• Make replicas always the same, thus the set of replicas appear to be a logical single machine

Replicated State Machine (Deterministic Automation)

Today when people talk about fault tolerance in distributed systems, they mean RSM techniques to keep system state safe under failures.

Determinism is essential for RSM

A deterministic state machine satisfies:

same state + same input -> same next state and output

A state machine’s outputs must be determined by its request sequence, rather than timing or unrelated external activity.

Each replica (state machine) receives the same set of operations and executes them in the same order

• How do we maintain the same set of operation for every replica?
• How can we ensure each replica executes them in the same order?
• A bunch of techniques exist; we start with a simple one: Primary Backup

Primary Backup

A replica group contains (N) replicas, which at any moment:

• exactly one is the primary;
• the others are backups.

Primary responsibilities:

• receives client requests;
• chooses their total order;
• assigns log positions;
• replicates the log;
• usually sends the client response.

Backup responsibilities:

• receives operations from the primary;
• stores them in the same log positions;
• executes committed operations in order, or saves them for later execution;
• takes over if the primary fails.

Normal Primary-Backup Operations

1. Primary gets operations
2. Primary orders ops as a sequence of totally ordered items into log
3. Replicates log to backups
4. Backup executes opts or just inserts into log
5. Primary gets ack from backup and then reply to client

Multicore execution can break determinism

The observation is:

• a primary may process requests concurrently on multiple cores;
• a backup may replay the log sequentially.

Suppose primary executes two operations that modify the same state:

A: x = x + 1
B: x = x * 2

The primary might physically interleave them in one way while the backup executes them in another.

Solution

Therefore, the primary must not merely log when requests arrived. It must log the logical order in which their effects committed:

log[10] = A
log[11] = B

The backup then reproduces that logical order.

Replication must capture the output of concurrency control, such as:

• the chosen serialization order;
• committed values;
• lock or transaction outcomes;
• nondeterministic decisions.

What if some operations are non-deterministic

Some operations do not produce results determined solely by state and input. For example:

• random()
• timeofday()

Solution

The solution is to replicate the decision. For example,

instead of letting every replica call the clock independently:

expire_session(timeofday())

the primary chooses a time and logs:

expire_session(chosen_time=14:03:17)

The backups replay the primary’s chosen value.

A good general rule is:

Perform nondeterminism once, convert its result into deterministic replicated input, and make every replica apply that result.

Failover Duplicate execution problem

Suppose the request is:

withdraw(account, $100)

The request commits, but the reply is lost.

The client retries:

withdraw(account, $100)

If the new primary treats it as a new request, the account loses $200.

Solution: Request Identifiers

Clients attach a unique identifier or monotonically increasing sequence number:

client = C7
request number = 42
operation = withdraw($100)

Replicas store a client table:

C7:
    highest completed request = 42
    cached reply = OK

When request 42 arrives again, the new primary:

1. recognizes it as a duplicate;
2. does not execute the withdrawal again;
3. resends the cached response.

This provides at-most-once execution.

Primary Failure Handling

Configurations, views, and epochs

A configuration identifies:

• the current primary;
• the backups;
• often a monotonically increasing configuration number.

It may be represented as:

configuration 8:
    primary = P
    backups = {B1, B2}

After failover:

configuration 9:
    primary = B1
    backups = {B2, P}

Other protocols call a configuration an:

• epoch;
• term;
• view.

The increasing number distinguishes messages from old and new leadership periods.

Atomic test-and-set

There is a shared storage containing the primary role.

A candidate executes an atomic test-and-set:

if primary_role is unclaimed:
    claim primary_role

Atomicity ensures that if multiple backups race, only one succeeds.