Fault Tolerance & Reliability

Before we can design for failure, we need a precise vocabulary for the kinds of failure that can occur. The distributed systems literature has settled on a hierarchy ordered by how difficult the failure is to handle.

Crash-stop (fail-stop) failures: A node simply stops executing and sends no further messages. This is the most benign failure mode and the one most OS-level software assumes. The key property is that a crashed node does not lie — it just disappears. If you can detect it (via heartbeats or timeouts), you can work around it.

Crash-recovery failures: A node crashes but may later restart, possibly with some state restored from durable storage. This is the model most real databases use. The complication is that a restarted node might have stale state and send messages consistent with an earlier view of the world.

Omission failures: A node is running but silently drops some messages — either on send or receive. Network packet loss falls in this category. The system appears partially connected in unpredictable ways.

Byzantine (arbitrary) failures: A node can do anything: send incorrect data, send different values to different peers, selectively delay messages, even actively collude with other Byzantine nodes. Byzantine failures model hardware corruption, software bugs that produce garbage output, and adversarial attackers. They are vastly harder to tolerate than crash failures.

The hierarchy matters because algorithms designed for one failure model may be completely broken under a stronger one. Paxos, for instance, is correct under crash-stop failures but can be subverted by a single Byzantine node. Practical Byzantine Fault Tolerance (PBFT) can handle Byzantine failures but requires 3f+1 nodes to tolerate f failures (versus 2f+1 for crash-stop).

A useful mental model: imagine a military unit. Crash-stop is a soldier who gets shot and falls silent. Omission is a radio operator who randomly drops transmissions. Byzantine is a traitor who sends false orders designed to maximize confusion.

The Two Generals' Problem illustrates the fundamental difficulty: two armies need to coordinate an attack over an unreliable messenger channel. It can be proved that no protocol guarantees both sides commit to the same action — acknowledgment of acknowledgment of acknowledgment ad infinitum never converges to certainty. This is why TCP three-way handshakes still don't give you absolute certainty of synchronized connection state.

# Failure model hierarchy
# (Ordered easiest → hardest to tolerate)

FAILURE_MODELS = {
    'crash_stop':    'Node halts, sends no more messages. Detectable via timeout.',
    'crash_recovery':'Node halts, may restart with partial state from disk.',
    'omission':      'Node runs but silently drops some send/receive messages.',
    'byzantine':     'Node may send arbitrary, malicious, or inconsistent messages.',
}

# Nodes needed to tolerate f failures:
# Crash-stop:  2f + 1 replicas (simple majority quorum)
# Byzantine:   3f + 1 replicas (PBFT / BFT consensus)
def min_replicas_needed(f: int, failure_model: str) -> int:
    if failure_model in ('crash_stop', 'crash_recovery'):
        return 2 * f + 1
    elif failure_model == 'byzantine':
        return 3 * f + 1
    raise ValueError('Unknown model')

The single most important idea in fault tolerance is redundancy — having more resources than strictly necessary so that failures in some subset do not cause system failure overall. Redundancy appears at every level of the hardware and software stack.

Hardware redundancy: RAID disk arrays distribute data (and optionally parity) across multiple disks. RAID-1 mirrors every write to two disks — you can lose any one disk and lose no data. RAID-5 uses distributed parity, tolerating one disk failure with less storage overhead. Modern servers have redundant power supplies, redundant NICs bonded together, and ECC RAM that detects and corrects single-bit memory errors.

Information redundancy: Error-correcting codes (ECC, Reed-Solomon) add redundant bits to data so that errors can be detected and corrected without retransmission. A Hamming(7,4) code encodes 4 data bits as 7 bits in a way that allows single-bit error correction. QR codes use Reed-Solomon codes so aggressively that up to 30% of the code can be destroyed and the data is still recoverable.

Time redundancy: Retransmitting a message after a timeout is time redundancy — the same operation is attempted multiple times. TCP's retransmission logic is the canonical example. Idempotency (making operations safe to repeat) is the critical property needed for time redundancy to be correct.

Process/service redundancy: Running multiple copies of a service (replicas) behind a load balancer. If one replica crashes, traffic is rerouted to the survivors. This is the standard deployment model for every large web service.

The key tension in all redundancy is cost vs. coverage. Triple-modular redundancy (TMR) — three copies vote on the answer — is used in aerospace because human lives depend on it, despite the 3x hardware cost. A startup might accept a 99.9% SLA (8.7 hours downtime/year) achievable with a single replicated pair rather than paying for 99.999% (5 minutes/year) that requires more complex quorum logic.

Quorums are the mathematical heart of redundancy reasoning. In a system with N replicas, a write quorum W and read quorum R must satisfy W + R > N to guarantee that every read sees at least one replica that participated in every write. This ensures reads never miss the latest write. The classic setting is N=3, W=2, R=2: any two replicas must agree on writes, and any two must be consulted on reads — tolerating one failure. Amazon Dynamo uses configurable N/W/R precisely for this trade-off.

# Quorum calculation for a replicated key-value store
# Given N replicas, find write/read quorum sizes

def valid_quorum(N: int, W: int, R: int) -> bool:
    """A valid quorum requires: W + R > N (read-write overlap)
    and W > N/2 (no two conflicting writes can both succeed)."""
    return (W + R > N) and (W > N // 2)

# Example: N=5, tolerate 2 failures
N = 5
configs = [
    (3, 3), # classic strict quorum
    (4, 2), # write-heavy: faster reads, slower writes  
    (2, 4), # read-heavy: faster writes, slower reads
    (1, 5), # async replication: fastest writes, all must read
]
for W, R in configs:
    ok = valid_quorum(N, W, R)
    print(f'W={W} R={R}: valid={ok}, tolerates {min(N-W, N-R)} failures')

# Output:
# W=3 R=3: valid=True, tolerates 2 failures
# W=4 R=2: valid=True, tolerates 1 failures  
# W=2 R=4: valid=False — two conflicting writes could both succeed
# W=1 R=5: valid=False — W is not a majority

When a process crashes and restarts, it needs to recover its state. The naive approach — always keeping all state in memory — means any crash loses all progress. Checkpointing is the discipline of periodically writing a consistent snapshot of state to durable storage so that recovery can start from a recent point rather than the beginning.

The key insight is the checkpoint interval trade-off: frequent checkpoints reduce recovery time but impose I/O overhead during normal operation; infrequent checkpoints are cheap but mean more work is lost and replayed on recovery.

Write-Ahead Logging (WAL) is the cornerstone technique in database fault tolerance. The protocol is simple but profound:

1. Before modifying any data, write a log record describing the modification to durable storage.
2. Only after the log record is safely on disk, apply the modification to the data.

This ordering guarantee (log before data) means that on crash recovery, you can inspect the log and determine exactly which operations were completed and which were interrupted mid-flight. Operations with complete log records can be redone (REDO recovery); operations that started but have no commit record can be undone (UNDO recovery).

The canonical WAL record contains: [LSN (log sequence number), transaction ID, page ID, before-image, after-image, operation type]. ARIES (Algorithm for Recovery and Isolation Exploiting Semantics), used in IBM DB2 and variants of SQL Server and PostgreSQL, is the gold standard WAL recovery algorithm. It makes three passes: analysis (find dirty pages and active transactions at crash), redo (replay history from the last checkpoint), and undo (roll back incomplete transactions).

Coordinated vs. uncoordinated checkpointing becomes critical in multi-process distributed systems. If each process checkpoints independently, you may capture an inconsistent global state — process A has checkpointed after sending a message to B, but B's checkpoint was taken before the message arrived. Recovery would then replay the send but not the receive, creating a phantom message. The Chandy-Lamport algorithm solves this by using marker messages to capture a consistent global snapshot without stopping the system — a beautiful and practically important result.

# Simplified Write-Ahead Log for a key-value store
import json, os

class WALStore:
    def __init__(self, log_path: str, data_path: str):
        self.log_path = log_path
        self.data_path = data_path
        self.data = {}
        self._recover()

    def _append_log(self, record: dict):
        """Durably append a log record BEFORE touching data."""
        with open(self.log_path, 'a') as f:
            f.write(json.dumps(record) + '\n')
            f.flush()
            os.fsync(f.fileno())  # Force to disk — critical!

    def put(self, key: str, value: str):
        # 1. Write log record FIRST
        self._append_log({'op': 'PUT', 'key': key, 'value': value})
        # 2. Only then update in-memory state
        self.data[key] = value

    def delete(self, key: str):
        self._append_log({'op': 'DEL', 'key': key})
        self.data.pop(key, None)

    def _recover(self):
        """On startup, replay the log to reconstruct state."""
        if not os.path.exists(self.log_path):
            return
        with open(self.log_path) as f:
            for line in f:
                rec = json.loads(line)
                if rec['op'] == 'PUT':
                    self.data[rec['key']] = rec['value']
                elif rec['op'] == 'DEL':
                    self.data.pop(rec['key'], None)

In 1985, Fischer, Lynch, and Paterson published a result that shocked the distributed systems community: no deterministic asynchronous distributed system can solve consensus in the presence of even a single crash failure. This is FLP impossibility, one of the most important theoretical results in computer science.

What it means precisely: Consensus requires every non-faulty process to eventually decide a value (liveness), with all deciding processes choosing the same value (agreement), and the chosen value being one that was proposed (validity). FLP proves that in a fully asynchronous model — where message delays are unbounded and you cannot distinguish a slow node from a dead one — you cannot simultaneously guarantee all three properties if any one process might crash.

The proof intuition: The proof constructs a bivalent initial configuration — a starting state from which the protocol could reach either decision value 0 or 1. It then shows that the adversary (who controls message scheduling) can always keep the system in a bivalent state by carefully delaying messages, preventing the system from ever committing to a decision. Specifically, starting from any bivalent configuration, there always exists a sequence of events the adversary can arrange to reach another bivalent configuration, so the system never terminates with a decision.

Why it matters for practice: FLP doesn't say consensus is impossible — it says you must give up something. Real systems escape FLP by relaxing one of the three properties:

• Relax asynchrony (introduce timeouts / failure detectors): Paxos and Raft do this. They assume a partially synchronous model and use leader timeouts to detect failures.
• Relax determinism: Randomized protocols (Ben-Or's algorithm) can achieve consensus with high probability by using random coin flips that the adversary cannot predict.
• Relax liveness under partition: The CAP theorem formalizes this trade-off more broadly.

Understanding FLP transforms how you think about systems that claim to solve consensus: they must have hidden assumptions about timing, randomness, or failure detection. When a vendor says their system is "always consistent and always available" in the face of network partitions, FLP tells you they are lying.

# FLP impossibility cannot be implemented away — this shows
# why purely timeout-free consensus hangs under adversarial scheduling.

# The crux: in a fully async model, a 'slow' node is
# indistinguishable from a 'dead' one.

# Suppose two nodes must agree on 0 or 1.
# Node A proposes 0. Node B proposes 1.
# Messages may be delayed arbitrarily.

# Naive protocol:
def naive_consensus(my_proposal, recv_proposal):
    """Wait for the other side's proposal, return majority."""
    # In an async model, recv_proposal may NEVER arrive if the
    # sender crashed — or it might arrive in 1 second or 1 hour.
    # There is no correct timeout value. If we wait forever,
    # we violate liveness. If we timeout, we may decide wrong.
    if recv_proposal is not None:
        return min(my_proposal, recv_proposal)  # arbitrary tie-break
    else:
        # Did the other node crash, or is it just slow?
        # We CANNOT know in a purely async model.
        raise RuntimeError('FLP: cannot decide without timing assumption')

# Real systems escape FLP via partial synchrony (timeouts)
# or randomization — never by ignoring the theorem.

Many fault-tolerant protocols require exactly one node to act as coordinator or leader at any time: a primary that accepts writes, a scheduler that assigns tasks, a lock manager. Leader election is the problem of choosing one node from a group when any node may have failed.

The Bully Algorithm (Garcia-Molina 1982) is the textbook approach for synchronous networks. Nodes have unique IDs. Election is triggered when a node detects the leader is unresponsive:

1. The detecting node broadcasts an ELECTION message to all nodes with higher IDs.
2. Any higher-ID node responds with OK and starts its own election.
3. A node that receives no OK within a timeout declares itself leader and broadcasts COORDINATOR.

The "bully" name comes from higher-ID nodes bullying out lower-ID contenders. Time complexity: O(n²) messages. The weakness is that the node with the highest ID always wins, even if it has the highest latency — ID doesn't capture capability.

Raft (Ongaro & Ousterhout, 2014) is a consensus algorithm designed explicitly for understandability as an alternative to Paxos. It decomposes consensus into three subproblems:

1. Leader election: Nodes start as followers. If a follower hears no heartbeat from a leader within an election timeout (randomized, e.g., 150-300ms), it becomes a candidate, increments its term number, and requests votes. A candidate wins with a majority of votes.
2. Log replication: The leader accepts client writes, appends them to its log, and replicates to followers. An entry is committed once a majority acknowledges it.
3. Safety: Raft guarantees that committed entries are never lost via its Log Matching Property and Leader Completeness Property — a leader always has all committed entries from previous terms.

Paxos (Lamport 1989/1998) is the original landmark consensus algorithm. It has two phases:

• Prepare/Promise: A proposer picks a proposal number n, sends PREPARE(n) to a quorum. Acceptors promise not to accept proposals < n, and report the highest proposal they've already accepted.
• Accept/Accepted: The proposer sends ACCEPT(n, v) to the quorum with value v (either its own or the highest-numbered value it received in Promise responses). Acceptors accept unless they've promised something higher.

Paxos is notoriously difficult to implement correctly at production scale (Multi-Paxos, leader leases, log compaction). Raft was designed to be easier to implement correctly, and the majority of new distributed systems (etcd, CockroachDB, TiKV) use Raft.

# Simplified Raft leader election state machine
from enum import Enum
import random, time

class State(Enum):
    FOLLOWER = 'follower'
    CANDIDATE = 'candidate'
    LEADER = 'leader'

class RaftNode:
    def __init__(self, node_id: int, peers: list):
        self.id = node_id
        self.peers = peers
        self.state = State.FOLLOWER
        self.current_term = 0
        self.voted_for = None        # who we voted for this term
        self.votes_received = set()
        # Randomized timeout avoids split votes
        self.election_timeout = random.uniform(0.15, 0.30)  # seconds
        self.last_heartbeat = time.time()

    def on_heartbeat_timeout(self):
        """Called when election_timeout elapses with no leader heartbeat."""
        self.state = State.CANDIDATE
        self.current_term += 1
        self.voted_for = self.id       # vote for self
        self.votes_received = {self.id}
        # Broadcast RequestVote to all peers
        for peer in self.peers:
            peer.on_request_vote(self.id, self.current_term)

    def on_request_vote(self, candidate_id: int, term: int):
        """Grant vote if term is new and we haven't voted."""
        if term > self.current_term:
            self.current_term = term
            self.state = State.FOLLOWER
            self.voted_for = None
        if self.voted_for is None:
            self.voted_for = candidate_id
            return True   # grant vote
        return False

    def on_vote_received(self, from_id: int):
        if self.state != State.CANDIDATE:
            return
        self.votes_received.add(from_id)
        majority = (len(self.peers) + 1) // 2 + 1
        if len(self.votes_received) >= majority:
            self.state = State.LEADER
            print(f'Node {self.id} elected leader for term {self.current_term}')

Traditional distributed protocols often rely on a central coordinator or structured broadcast trees. These introduce single points of failure. Gossip protocols (also called epidemic protocols) achieve reliable information dissemination in a completely decentralized, fault-tolerant way, inspired by how rumors spread in a social network.

The core loop is simple: periodically, each node selects a random peer and exchanges information (state, membership updates, or failure notifications). This seemingly naive approach has remarkable theoretical properties:

• Convergence speed: A gossip round takes O(log N) steps to propagate information to all N nodes, because each round doubles the number of informed nodes on average (like binary doubling).
• Robustness: Even if half the nodes fail simultaneously, gossip still converges. There is no single bottleneck.
• Scalability: Each node sends a constant number of messages per round regardless of cluster size.

Push vs. Pull vs. Push-Pull gossip: In push gossip, an informed node sends to a random peer. In pull gossip, a node asks a random peer for updates it might have missed. Push-pull combines both and achieves faster convergence: in the early rounds (few informed nodes), push is efficient; in later rounds (many informed nodes), pull cleans up stragglers.

Membership and failure detection with SWIM: The Scalable Weakly-consistent Infection-style Membership (SWIM) protocol uses gossip to maintain a membership list and detect failures without a central heartbeat service. Nodes periodically ping random members. If a ping times out, the node asks k other members to probe the suspect (indirect probing). Only after multiple independent failures to reach a node is it declared dead. This dramatically reduces false positives compared to simple timeout-based detection. HashiCorp Consul, Amazon DynamoDB's failure detection, and Kubernetes Gossip (used in some clusters) are built on SWIM or variants.

Convergent replicated data types (CRDTs) work beautifully with gossip: each node maintains a locally-updateable data structure that can be merged with any other node's version, and the gossip protocol ensures all nodes eventually converge to the same state without coordination. AWS DynamoDB's shopping cart (eventually consistent) was an early practical application.

import random
from typing import Dict, Set

class GossipNode:
    """Push-pull gossip for propagating a key-value update."""
    def __init__(self, node_id: int):
        self.id = node_id
        self.state: Dict[str, int] = {}   # key -> version
        self.peers: list = []             # set externally

    def update(self, key: str, version: int):
        """Local write — node learns about this update first."""
        self.state[key] = max(self.state.get(key, -1), version)

    def gossip_round(self):
        """Push-pull: exchange full state digest with one random peer."""
        if not self.peers:
            return
        peer = random.choice(self.peers)
        # PUSH: send what we know
        # PULL: receive what they know
        # In practice this is a network RPC; here we simulate directly:
        merged = self._merge(self.state, peer.state)
        self.state = merged
        peer.state = merged

    @staticmethod
    def _merge(a: Dict, b: Dict) -> Dict:
        """Last-write-wins merge: take max version for each key."""
        keys = set(a) | set(b)
        return {k: max(a.get(k, -1), b.get(k, -1)) for k in keys}

# Simulate convergence across 5 nodes
nodes = [GossipNode(i) for i in range(5)]
for n in nodes:
    n.peers = [p for p in nodes if p.id != n.id]

nodes[0].update('config_version', 42)   # only node 0 knows

for round_num in range(1, 8):
    for node in nodes:
        node.gossip_round()
    informed = sum(1 for n in nodes if n.state.get('config_version') == 42)
    print(f'Round {round_num}: {informed}/5 nodes informed')
# Typically converges in 3-4 rounds (O(log 5) ≈ 2.3)

Brewer's CAP theorem (2000, formally proved by Gilbert & Lynch 2002) states that a distributed data store can provide at most two of three properties simultaneously:

• Consistency: Every read receives the most recent write or an error (linearizability).
• Availability: Every request receives a response (not an error), though it might not be the most recent data.
• Partition tolerance: The system continues operating despite network partitions (some nodes cannot communicate).

The critical insight is that partition tolerance is not optional in a real distributed system — network partitions happen whether you want them to or not. Therefore, the real trade-off is between C and A during a partition: do you refuse to serve stale reads (choose C), or serve possibly stale data rather than returning an error (choose A)?

CA systems (sacrifice P): traditional single-node relational databases. MySQL with a single master: linearizable, always available, but if the network splits, the system as a whole stops (the partitioned side refuses writes). Not truly distributed.

CP systems (sacrifice A during partition): HBase, Zookeeper, etcd, Google Spanner. During a partition, they refuse reads/writes to the minority partition to prevent stale data. Correct, but unavailable.

AP systems (sacrifice C during partition): Amazon Dynamo, Cassandra, CouchDB. During a partition, all nodes serve reads and accept writes, possibly diverging. They converge after the partition heals via anti-entropy (gossip + conflict resolution).

PACELC is a refinement by Abadi (2012) that notes the C/A trade-off also exists in the common case (no partition): even without a partition, there is a trade-off between Latency and Consistency — achieving linearizability requires coordination (round trips) that increases latency. Most real systems expose tunable consistency levels (Cassandra's CONSISTENCY ONE/QUORUM/ALL) so operators can choose where on the spectrum they want to sit for each query.

# PACELC trade-off table for common databases
# P=partition present, E=else (no partition)
# C=consistency, A=availability, L=low latency

systems = {
    # name:              (partition choice,  normal choice)
    'Zookeeper/etcd':    ('CP: refuses writes on minority side',
                          'EC: linearizable reads add round-trip latency'),
    'Cassandra ONE':     ('AP: all nodes serve requests',
                          'EL: single-replica reads are fast but may be stale'),
    'Cassandra QUORUM':  ('CP: minority partition returns error',
                          'EC: quorum reads add latency but are consistent'),
    'Google Spanner':    ('CP: globally consistent via TrueTime',
                          'EC: external consistency, higher latency (~10ms)'),
    'Amazon DynamoDB':   ('AP by default (eventual)',
                          'EL: single-digit ms reads'),
    'MySQL (single)':    ('CA: not truly partition tolerant',
                          'EC: ACID, single-node latency'),
}
for system, (p_choice, e_choice) in systems.items():
    print(f'{system}:')
    print(f'  Partition: {p_choice}')
    print(f'  Normal:    {e_choice}')

Replication is how distributed systems achieve both fault tolerance (survive node failures) and higher availability (serve reads from multiple replicas). The choice of replication strategy determines the consistency guarantees, write latency, and failure recovery behavior.

Primary-Backup Replication: One replica is designated primary (or master/leader). All writes go to the primary, which replicates synchronously or asynchronously to backup replicas. On primary failure, a backup is promoted.

• Synchronous replication: Primary waits for acknowledgment from all (or a quorum of) backups before confirming the write to the client. No data loss on failover. Higher write latency.
• Asynchronous replication: Primary confirms immediately, backups catch up eventually. Low write latency. Risk of data loss if primary crashes before replicating recent writes (replication lag).

Chain Replication (van Renesse & Schneider 2004) arranges replicas in a linear chain. Writes enter at the head, propagate down the chain, and are acknowledged to the client by the tail. Reads are served only by the tail (always consistent). This achieves strong consistency with O(n) message chain cost per write, but excellent read performance and simplified failure recovery — just remove the failed node from the chain.

Multi-Paxos / Raft log replication: Rather than replicating a single value, production systems replicate a log of commands. Each log entry is a command to apply to a state machine. All replicas apply the same commands in the same order, so they stay in sync (the State Machine Replication (SMR) model). This is how etcd, CockroachDB, and Google Chubby work. The key insight (Lamport's State Machine Replication) is that consensus on a sequence of commands is equivalent to sequential consensus on each individual command.

Split-brain is the most dangerous failure mode in replication: two nodes both believe they are the primary (e.g., after a network partition where each side's heartbeats to the other time out). Both accept writes that diverge. The solution is fencing tokens: a monotonically increasing number issued by the old primary on retirement, passed with every write to the storage backend. The backend rejects writes with stale tokens, ensuring only the true current primary can write.

# Chain Replication: write flows head→tail, reads from tail only

class ChainNode:
    def __init__(self, node_id, successor=None):
        self.id = node_id
        self.store = {}
        self.successor = successor  # next node in chain

    def write(self, key, value, ack_callback):
        """Propagate write down the chain."""
        self.store[key] = value
        if self.successor:
            # Forward to next node in chain
            self.successor.write(key, value, ack_callback)
        else:
            # We are the TAIL — ack to client
            ack_callback(key, value, success=True)

    def read(self, key):
        """Only the TAIL serves reads — always consistent."""
        # In real chain replication, only the tail node should
        # have this method exposed to clients.
        return self.store.get(key)

# Build a 3-node chain: head -> middle -> tail
tail   = ChainNode('tail')
middle = ChainNode('middle', successor=tail)
head   = ChainNode('head',   successor=middle)

acks = []
head.write('x', 42, lambda k, v, s: acks.append((k, v, s)))
print(f'Tail value: {tail.read("x")}')  # 42
print(f'Acks: {acks}')                  # [('x', 42, True)]

A system cannot recover from failures it cannot detect. Failure detection — determining that a remote process has crashed — is one of the most subtle problems in distributed systems, because of the fundamental indistinguishability between a slow node and a dead node in an asynchronous system.

Heartbeats: The simplest approach. Each process periodically sends a "I'm alive" message. If a peer hasn't sent a heartbeat within a timeout window, it's declared failed. The challenge: what timeout to use?

• Too short: false positives — a temporarily slow or network-congested node is declared dead, triggering unnecessary failovers and potentially split-brain.
• Too long: slow detection — a truly dead node is not detected for a long time, leaving the system degraded.

The Φ (phi) Accrual Failure Detector (used in Cassandra and Akka) solves this elegantly. Instead of a binary dead/alive judgment, it outputs a continuous suspicion level φ. φ is computed from the distribution of historical inter-heartbeat intervals. If heartbeats arrive every ~100ms normally, and the last one was 500ms ago, φ might be 3.0 (low suspicion). If it's been 2 seconds, φ might be 12.0 (high suspicion). The application sets its own threshold (e.g., declare failed when φ > 8), balancing false positives vs. detection latency based on its risk tolerance.

Perfect failure detectors are impossible in an asynchronous system (by FLP). In practice, we use eventually perfect failure detectors: they may initially make mistakes (declaring live nodes dead or vice versa) but eventually are correct — a dead node is eventually permanently suspected, and a live node is eventually permanently trusted.

Lease-based failure detection is an important pattern for preventing split-brain. A lease is a time-bounded grant of authority (e.g., "you are the primary until time T + δ"). A node must renew its lease before it expires. If it cannot (because it's partitioned from the lease grantor), it stops acting as primary rather than risk diverging. This is the mechanism behind Google Chubby's master election and Kubernetes leader election via etcd.

import math, collections, time

class PhiAccrualDetector:
    """Simplified phi accrual failure detector.
    Maintains a sliding window of inter-arrival times.
    """
    def __init__(self, window_size: int = 200, threshold: float = 8.0):
        self.intervals = collections.deque(maxlen=window_size)
        self.last_heartbeat_time = None
        self.threshold = threshold

    def heartbeat(self):
        """Call this when a heartbeat is received."""
        now = time.monotonic()
        if self.last_heartbeat_time is not None:
            interval = now - self.last_heartbeat_time
            self.intervals.append(interval)
        self.last_heartbeat_time = now

    def phi(self) -> float:
        """Compute suspicion level. Higher = more suspicious."""
        if len(self.intervals) < 2 or self.last_heartbeat_time is None:
            return 0.0
        elapsed = time.monotonic() - self.last_heartbeat_time
        mean = sum(self.intervals) / len(self.intervals)
        # Approximation using exponential distribution
        # phi = elapsed / mean * log(10) (simplified)
        if mean == 0:
            return float('inf')
        return (elapsed / mean) * math.log(10)

    def is_alive(self) -> bool:
        return self.phi() < self.threshold

The most important design philosophy in fault-tolerant systems is graceful degradation: when failures occur, the system should reduce its capabilities in proportion to the failure, rather than failing completely. A system that degrades gracefully is said to be resilient; one that fails catastrophically on partial failure is brittle.

Bulkheads (from ship design): isolate components so that a failure in one doesn't flood the others. In software, this means separate thread pools for different downstream services (if one service is slow and backs up its thread pool, it can't steal threads from other services), separate process groups, circuit breakers at service boundaries.

Circuit breakers (Nygard's Release It!): wrap calls to external services. A circuit breaker has three states:

• Closed (normal): requests pass through.
• Open (failing): after a threshold of failures, the breaker opens and all calls immediately return an error or cached response, without actually trying the failing service. This prevents cascading failures and gives the downstream service time to recover.
• Half-open (probing): after a timeout, let a few requests through. If they succeed, close the breaker. If not, reopen it.

Timeout and retry discipline: Every external call must have a timeout. Retry with exponential backoff and jitter (randomized delay) avoids thundering herds — the situation where all clients retry simultaneously after a failure, overwhelming the recovering service. AWS's SDK retry logic is the canonical implementation.

Idempotency keys: For operations with side effects (payments, sends), assign each request a unique idempotency key. The server stores processed keys. On retry, if the key is already in the store, return the previous result without reprocessing. This makes retries safe.

Chaos engineering (popularized by Netflix's Chaos Monkey): proactively inject failures in production to find hidden dependencies before they cause real outages. Netflix's Chaos Monkey randomly terminates EC2 instances; Chaos Kong brings down an entire AWS availability zone. Systems not tested for failure quietly accumulate fragility that only manifests in the worst possible moment.

# Circuit breaker implementation
import time
from enum import Enum

class CBState(Enum):
    CLOSED = 'closed'       # normal
    OPEN = 'open'           # failing fast
    HALF_OPEN = 'half_open' # testing recovery

class CircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=30):
        self.state = CBState.CLOSED
        self.failure_count = 0
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.last_failure_time = None

    def call(self, func, *args, **kwargs):
        if self.state == CBState.OPEN:
            # Check if recovery timeout elapsed
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = CBState.HALF_OPEN
            else:
                raise RuntimeError('Circuit is OPEN — failing fast')
        try:
            result = func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise

    def _on_success(self):
        self.failure_count = 0
        self.state = CBState.CLOSED

    def _on_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.failure_count >= self.failure_threshold:
            self.state = CBState.OPEN
            print(f'Circuit OPENED after {self.failure_count} failures')

When a transaction spans multiple nodes (e.g., debit account A on server 1 and credit account B on server 2), all nodes must agree to commit or all must abort — partial commits leave the system in an inconsistent state. Two-Phase Commit (2PC) is the classical protocol for atomic distributed commitment.

Phase 1 — Prepare (Voting): The coordinator sends a PREPARE message to all participants. Each participant:

1. Writes the transaction's undo and redo information to its log.
2. Acquires all needed locks.
3. Responds with VOTE-COMMIT (if ready) or VOTE-ABORT (if not).

Phase 2 — Commit or Abort: If all participants voted COMMIT, the coordinator logs COMMIT and sends COMMIT to all. If any participant voted ABORT, the coordinator logs ABORT and sends ABORT to all. Participants then apply or undo the transaction and release locks.

2PC is a blocking protocol — this is its fundamental weakness. If the coordinator crashes after Phase 1 but before sending Phase 2, participants that voted COMMIT are stuck holding locks and cannot decide. They cannot commit (coordinator might have decided ABORT) and cannot abort (coordinator might have decided COMMIT). This uncertainty window lasts until the coordinator recovers. In practice, coordinators use WAL to recover, but the protocol can block for arbitrarily long.

Three-Phase Commit (3PC) adds a PRE-COMMIT phase to eliminate the blocking window in certain failure scenarios, at the cost of additional round trips. However, 3PC is not safe under network partitions and is rarely used in practice.

Saga pattern (Hector Garcia-Molina 1987, popularized in microservices): instead of a distributed transaction, decompose a long-running business operation into a sequence of local transactions, each with a compensating transaction that undoes it. If step k fails, run compensations for steps k-1, k-2, ..., 1 in reverse. Sagas don't require holding locks across services and are the standard pattern for distributed business workflows in microservices architectures.

# Two-Phase Commit protocol sketch

class Participant:
    def __init__(self, name):
        self.name = name
        self.locked = False
        self.prepared = False

    def prepare(self, txn) -> bool:
        """Phase 1: write to WAL, acquire locks, vote."""
        # Simulate resource check
        can_commit = True  # e.g., check constraints, acquire row locks
        if can_commit:
            self.prepared = True
            self.locked = True
            print(f'{self.name}: VOTE-COMMIT')
        else:
            print(f'{self.name}: VOTE-ABORT')
        return can_commit

    def commit(self):
        assert self.prepared, 'Must have prepared first!'
        self.locked = False  # release locks
        self.prepared = False
        print(f'{self.name}: COMMITTED')

    def abort(self):
        self.locked = False
        self.prepared = False
        print(f'{self.name}: ABORTED')

class Coordinator:
    def __init__(self, participants):
        self.participants = participants

    def run_transaction(self, txn):
        # Phase 1: collect votes
        votes = [p.prepare(txn) for p in self.participants]
        # Phase 2: commit or abort (must be durable logged first!)
        if all(votes):
            for p in self.participants:
                p.commit()
            return 'COMMITTED'
        else:
            for p in self.participants:
                p.abort()
            return 'ABORTED'