# Batch and Stream Processing

Data processing systems fall into three categories: services (online systems that handle requests), batch processing (offline systems that process large volumes of accumulated data), and stream processing (near-real-time systems that process data as it arrives). Understanding when and how to use batch and stream processing is essential for building data pipelines, analytics systems, and derived data stores.


## Table of Contents
1. [Batch Processing: MapReduce and Beyond](#batch-processing-mapreduce-and-beyond)
2. [Dataflow Engines: Beyond MapReduce](#dataflow-engines-beyond-mapreduce)
3. [Event Sourcing](#event-sourcing)
4. [Change Data Capture (CDC)](#change-data-capture-cdc)
5. [Stream-Table Duality](#stream-table-duality)
6. [Exactly-Once Semantics](#exactly-once-semantics)
7. [Time Windowing](#time-windowing)
8. [Architecture Patterns](#architecture-patterns)

---

## Batch Processing: MapReduce and Beyond

### The MapReduce Paradigm

MapReduce, popularized by Google's 2004 paper, processes large datasets by breaking computation into two phases:

1. **Map phase:** Read input data, extract key-value pairs. Each mapper processes a portion of the input independently.
2. **Shuffle phase:** Framework groups all values by key, distributing them to reducers.
3. **Reduce phase:** For each key, combine all values into a result.

```
Input: ["the cat sat on the mat"]

Map:
  "the" -> 1
  "cat" -> 1
  "sat" -> 1
  "on"  -> 1
  "the" -> 1
  "mat" -> 1

Shuffle: Group by key
  "the" -> [1, 1]
  "cat" -> [1]
  "sat" -> [1]
  "on"  -> [1]
  "mat" -> [1]

Reduce: Sum values
  "the" -> 2
  "cat" -> 1
  "sat" -> 1
  "on"  -> 1
  "mat" -> 1
```

### MapReduce Strengths

- **Horizontal scalability:** Add more machines to process larger datasets; the framework handles distribution
- **Fault tolerance:** If a mapper or reducer fails, the framework re-executes that task on another machine. Input data is immutable, so re-execution is safe.
- **Simplicity:** The programmer only writes map and reduce functions; the framework handles parallelism, distribution, and fault tolerance

### MapReduce Weaknesses

- **High latency:** Each MapReduce job reads from and writes to distributed storage (HDFS), adding significant I/O overhead
- **Chaining is awkward:** Complex computations require chaining multiple MapReduce jobs, each with its own read/write cycle
- **No iteration support:** Machine learning algorithms that iterate over data must launch a new MapReduce job for each iteration
- **Limited expressiveness:** Not all computations fit the map-reduce pattern naturally

---

## Dataflow Engines: Beyond MapReduce

### Apache Spark

Spark replaces MapReduce with a more general computation model based on Resilient Distributed Datasets (RDDs) and directed acyclic graphs (DAGs) of operators.

**Key improvements over MapReduce:**
- **In-memory processing:** Intermediate results stay in memory instead of being written to disk between stages
- **Arbitrary DAGs:** Computations can have multiple stages with various operators (map, filter, join, group, sort), not just map and reduce
- **Lazy evaluation:** Spark builds a computation plan before executing, enabling the optimizer to eliminate unnecessary steps
- **Iterative algorithms:** RDDs can be cached in memory and reused across iterations

```python
# Spark: Word count
text_file = spark.read.text("hdfs://input/")
word_counts = (text_file
    .select(explode(split(col("value"), " ")).alias("word"))
    .groupBy("word")
    .count()
    .orderBy(desc("count")))
word_counts.write.parquet("hdfs://output/")
```

### Apache Flink

Flink treats batch processing as a special case of stream processing (a bounded stream). Its architecture is stream-first.

**Key features:**
- **True streaming:** Processes events one at a time (not micro-batches like Spark Streaming)
- **Event-time processing:** Handles out-of-order events based on when they occurred, not when they arrived
- **Exactly-once semantics:** Provides exactly-once processing guarantees through checkpointing
- **Savepoints:** Snapshot the entire pipeline state for upgrades, scaling, or debugging

### Comparison

| Feature | MapReduce | Spark | Flink |
|---------|-----------|-------|-------|
| **Processing model** | Batch only | Batch + micro-batch streaming | Batch + true streaming |
| **Intermediate storage** | Disk (HDFS) | Memory (spills to disk) | Memory (checkpoints to disk) |
| **Latency** | Minutes to hours | Seconds to minutes | Milliseconds to seconds |
| **Fault tolerance** | Re-execute failed tasks | Recompute lost RDD partitions | Checkpoint-based recovery |
| **Best for** | Very large batch jobs | Interactive analytics, ML | Real-time streaming, event-time processing |

---

## Event Sourcing

### Concept

Instead of storing the current state of an entity, store every state change as an immutable event. The current state is derived by replaying all events.

```
Traditional (mutable state):
  Account { id: 1, balance: 150 }

Event sourcing (immutable log):
  AccountCreated  { id: 1, balance: 0 }
  MoneyDeposited  { id: 1, amount: 200 }
  MoneyWithdrawn  { id: 1, amount: 50 }

  Current state = replay events: 0 + 200 - 50 = 150
```

### Benefits

- **Complete audit trail:** Every change is recorded with timestamp, actor, and context
- **Temporal queries:** "What was the balance on January 15?" Replay events up to that date
- **Event replay:** Rebuild read models, fix bugs by replaying with corrected logic, build new views from historical events
- **Debugging:** Reproduce any state by replaying the exact sequence of events
- **Decoupling:** Event producers and consumers can evolve independently

### Challenges

- **Event schema evolution:** Once events are stored, changing their schema is hard; use versioned event schemas
- **Eventual consistency:** Read models derived from events may lag behind the event log
- **Storage growth:** The event log grows forever; compaction or snapshotting is needed for old events
- **Complexity:** Building and maintaining projections (read models) adds architectural complexity

### Event Store Implementations

| Technology | Type | Key Feature |
|-----------|------|-------------|
| **EventStoreDB** | Purpose-built event store | Projections, subscriptions, optimistic concurrency |
| **Apache Kafka** | Distributed log | High throughput, log compaction, exactly-once semantics |
| **PostgreSQL** | Relational DB as event store | ACID transactions on event writes; LISTEN/NOTIFY for subscribers |
| **DynamoDB Streams** | Change stream | Automatic change capture from DynamoDB tables |

---

## Change Data Capture (CDC)

### Concept

CDC observes all writes to a database and extracts them as a stream of change events that can be consumed by other systems. This keeps derived data stores (search indexes, caches, analytics databases) in sync with the source of truth.

```
Application -> PostgreSQL (source of truth)
                   |
                   v (CDC)
              Kafka topic
              /    |    \
             v     v     v
    Elasticsearch  Redis  Data Warehouse
    (search)     (cache)  (analytics)
```

### CDC Implementation Approaches

| Approach | How It Works | Trade-off |
|----------|-------------|-----------|
| **Log-based (WAL parsing)** | Read the database's write-ahead log and extract changes | Most reliable; low overhead; captures all changes including those from direct SQL |
| **Trigger-based** | Database triggers write changes to an outbox table | Works with any database; adds write overhead; may miss changes from bulk operations |
| **Polling-based** | Periodically query for changed rows (using updated_at timestamp) | Simplest; misses deletes; can miss rapid changes between polls; adds query load |
| **Application-level** | Application explicitly writes events when modifying data | Full control; risk of forgetting to emit events; dual-write problem |

### CDC Tools

| Tool | Source Databases | Sink | Key Feature |
|------|-----------------|------|-------------|
| **Debezium** | PostgreSQL, MySQL, MongoDB, SQL Server, Oracle | Kafka | WAL-based; exactly-once; schema registry integration |
| **Maxwell** | MySQL only | Kafka, RabbitMQ, Redis | Lightweight; MySQL binlog parsing |
| **AWS DMS** | Most databases | Kafka, S3, Redshift, DynamoDB | Managed service; heterogeneous migration |
| **Fivetran/Airbyte** | Many sources | Data warehouses | Managed ELT platforms with CDC connectors |

### The Dual-Write Problem

A common anti-pattern is writing to two systems directly:

```
Application -> writes to PostgreSQL
           -> writes to Elasticsearch

Problem: If the Elasticsearch write fails after the PostgreSQL write succeeds,
the systems are now inconsistent. Retrying may cause duplicates.
```

**Solution:** Write to one system (PostgreSQL) and use CDC to propagate to the other. The CDC pipeline handles retries, ordering, and exactly-once delivery.

---

## Stream-Table Duality

### The Core Insight

A stream and a table are two sides of the same coin:

- **A stream is the changelog of a table.** If you record every INSERT, UPDATE, and DELETE to a table, that sequence of changes is a stream.
- **A table is the materialized state of a stream.** If you replay a stream of changes from the beginning, applying each change in order, you get the current table state.

```
Stream:
  INSERT user {id: 1, name: "Alice"}
  INSERT user {id: 2, name: "Bob"}
  UPDATE user {id: 1, name: "Alice Chen"}
  DELETE user {id: 2}

Table (materialized from stream):
  | id | name        |
  |----|-------------|
  | 1  | Alice Chen  |
```

### Practical Application: Kafka Log Compaction

Kafka's log compaction feature retains only the latest value for each key, effectively converting a stream into a table snapshot:

```
Before compaction:
  key=1, value="Alice"
  key=2, value="Bob"
  key=1, value="Alice Chen"
  key=2, value=null (tombstone)

After compaction:
  key=1, value="Alice Chen"
  (key=2 is deleted because value is null)
```

This allows a new consumer to read the compacted log and reconstruct the full current state without replaying the entire history.

---

## Exactly-Once Semantics

### The Challenge

In distributed systems, messages can be lost, duplicated, or reordered. "Exactly-once" means that the effect of processing each message is reflected exactly once in the output, even in the presence of failures.

### Achieving Exactly-Once

True exactly-once requires coordination between the messaging system and the processing logic:

| Approach | How It Works | Example |
|----------|-------------|---------|
| **Idempotent operations** | Design operations so that applying them multiple times has the same effect as applying once | `SET balance = 100` is idempotent; `INCREMENT balance BY 10` is not |
| **Transactional output** | Write output and update consumer offset in a single atomic transaction | Kafka Streams: transactional producer commits output records and consumer offsets together |
| **Deduplication** | Assign a unique ID to each message; recipient ignores messages it has already processed | Store processed message IDs in a set; check before processing |
| **Checkpointing** | Periodically save processing state; on failure, resume from last checkpoint | Flink savepoints: snapshot operator state and input positions |

---

## Time Windowing

### Why Windowing Matters

Unbounded streams have no natural "end," so you can't wait for all data before computing an aggregate. Windowing divides the stream into finite chunks for aggregation.

### Window Types

| Window Type | How It Works | Use Case |
|------------|-------------|----------|
| **Tumbling** | Fixed-size, non-overlapping windows (e.g., every 5 minutes) | Hourly metrics, daily summaries |
| **Hopping** | Fixed-size windows that overlap (e.g., 10-minute windows every 5 minutes) | Smoothed averages, sliding computations |
| **Session** | Variable-size windows based on activity gaps (e.g., a session ends after 30 minutes of inactivity) | User session analytics, click streams |
| **Global** | A single window for the entire stream | Running totals, all-time aggregates |

### Handling Late Events

Events may arrive after their window has closed (due to network delays, buffering, or clock skew):

- **Watermarks:** A timestamp that says "I believe all events before this time have arrived." Events arriving after the watermark are considered late.
- **Allowed lateness:** Accept late events up to a threshold (e.g., 1 hour after window closes), updating the window's result
- **Side outputs:** Route late events to a separate output for manual or delayed processing

```
Window: 10:00 - 10:05
Watermark: 10:06 (all events before 10:06 expected)
Allowed lateness: 1 hour

Event at 10:03 arriving at 10:07: accepted (within allowed lateness)
Event at 10:01 arriving at 11:30: discarded or sent to side output
```

---

## Architecture Patterns

### Lambda Architecture

Run both batch and stream processing pipelines in parallel:

```
Raw Data -> Batch Layer (Spark) -> Batch Views
         -> Speed Layer (Flink) -> Real-time Views

Query: Merge batch views + real-time views for complete result
```

**Pros:** Batch provides correctness; stream provides speed.
**Cons:** Maintaining two pipelines with the same logic is expensive and error-prone; results may differ between batch and stream.

### Kappa Architecture

Use a single stream processing pipeline for everything. Reprocess historical data by replaying the event log:

```
Event Log (Kafka) -> Stream Processor (Flink) -> Derived Views

Reprocessing: Start a new consumer from the beginning of the log
```

**Pros:** Single codebase; simpler architecture; easier to reason about.
**Cons:** Reprocessing large histories can be slow; requires a durable, replayable log (Kafka with long retention).

### Choosing Between Lambda and Kappa

Use **Lambda** when:
- Exact correctness is required and stream processing approximations are unacceptable
- You have existing batch infrastructure and are adding streaming incrementally

Use **Kappa** when:
- Your stream processing framework provides exactly-once guarantees
- You can afford to reprocess from the log when logic changes
- Simplicity and maintainability are priorities

# Data Models and Query Languages

Choosing a data model is the most consequential architectural decision in an application. The data model shapes not only how data is stored, but how you think about the problem domain, what queries are natural, and how the system evolves over time.

## The Relational Model

### When Relational Excels

The relational model, formalized by Edgar Codd in 1970, represents data as tables (relations) of rows (tuples) with typed columns (attributes). Its strength lies in:

- **Many-to-many relationships:** Foreign keys and joins make it natural to represent complex relationships without data duplication
- **Ad-hoc queries:** SQL's declarative nature lets you ask any question without pre-planned access paths
- **Referential integrity:** Foreign key constraints enforce that references point to existing records
- **Transaction support:** ACID transactions with mature isolation levels are standard
- **Schema enforcement (schema-on-write):** The database rejects data that doesn't conform to the schema, catching errors early

### Relational Anti-Patterns

The relational model struggles with:

- **Object-relational impedance mismatch:** Application objects don't map cleanly to flat tables; ORMs paper over this but add complexity
- **Deeply nested or tree-structured data:** Representing a resume with multiple jobs, each with multiple projects, each with multiple technologies requires many joins
- **Schema rigidity:** Adding a column to a table with billions of rows can be operationally expensive (though many databases now support instant `ADD COLUMN`)
- **Horizontal scaling:** Distributing joins across nodes is fundamentally hard

### SQL as a Query Language

SQL is declarative: you specify what you want, not how to get it. The optimizer chooses the execution plan, which means:

- Query performance can improve without changing application code (when the optimizer or indexes improve)
- Complex queries are concise compared to imperative alternatives
- The optimizer can parallelize, reorder joins, and choose index strategies

```sql
-- Find all users who placed an order in the last 30 days
-- and have a shipping address in California
SELECT DISTINCT u.name, u.email
FROM users u
JOIN orders o ON u.id = o.user_id
JOIN addresses a ON u.id = a.user_id
WHERE o.created_at > NOW() - INTERVAL '30 days'
  AND a.state = 'CA'
  AND a.type = 'shipping';
```

This query would require nested loops, hash maps, and set operations in imperative code. SQL expresses the intent in six lines.

---

## The Document Model

### When Document Excels

Document databases (MongoDB, CouchDB, Firestore) store data as self-contained documents, typically JSON or BSON:

- **One-to-many relationships:** When a parent entity contains a list of child entities that are always accessed together, a document model avoids joins entirely
- **Data locality:** Reading a single document retrieves all related data in one disk seek, improving read performance for aggregate access
- **Schema flexibility (schema-on-read):** Each document can have a different structure; the application interprets the schema at read time
- **Natural fit for aggregates:** Domain-driven design aggregates map directly to documents

### Document Example

```json
{
  "user_id": "u-4829",
  "name": "Alice Chen",
  "email": "alice@example.com",
  "addresses": [
    {"type": "home", "city": "San Francisco", "state": "CA"},
    {"type": "work", "city": "Palo Alto", "state": "CA"}
  ],
  "orders": [
    {
      "order_id": "o-1001",
      "items": [
        {"product": "Keyboard", "qty": 1, "price": 89.99},
        {"product": "Mouse", "qty": 2, "price": 29.99}
      ],
      "total": 149.97
    }
  ]
}
```

Everything about a user is in one document. No joins needed for the common access pattern of "show me everything about this user."

### Document Anti-Patterns

- **Many-to-many relationships:** Without joins, you must denormalize (duplicate data) or perform multiple queries and join in application code
- **Cross-document references:** If order items reference a shared product catalog, updating a product name requires updating every document that embeds it
- **Large documents:** Documents that grow unboundedly (e.g., an array of all user events) cause write amplification because the entire document must be rewritten
- **Deep nesting:** Querying deeply nested fields is possible but awkward; updating nested fields requires careful path expressions

---

## The Graph Model

### When Graph Excels

Graph databases (Neo4j, Amazon Neptune, JanusGraph) represent data as nodes (entities) and edges (relationships):

- **Highly interconnected data:** Social networks, knowledge graphs, fraud detection, recommendation engines
- **Recursive traversals:** "Find all people within 3 degrees of connection" is natural in graph query languages but requires recursive CTEs or multiple joins in SQL
- **Heterogeneous data:** Nodes and edges can have different types and properties without schema changes
- **Path analysis:** Shortest path, centrality, community detection algorithms are built into graph engines

### Graph Query Example (Cypher)

```cypher
// Find mutual friends who live in the same city
MATCH (alice:Person {name: 'Alice'})-[:FRIENDS_WITH]->(mutual)<-[:FRIENDS_WITH]-(bob:Person {name: 'Bob'})
WHERE mutual.city = alice.city
RETURN mutual.name, mutual.city
```

The equivalent SQL query would require self-joins and subqueries that obscure the intent.

### Graph Anti-Patterns

- **Simple CRUD operations:** Graph databases add overhead for straightforward create/read/update/delete without relationship traversals
- **Aggregation-heavy analytics:** Summing, counting, and grouping are more natural in SQL or column stores
- **Write-heavy workloads:** Graph index structures can be slower for bulk ingestion compared to LSM-tree stores

---

## Schema-on-Write vs. Schema-on-Read

| Aspect | Schema-on-Write (Relational) | Schema-on-Read (Document) |
|--------|------------------------------|---------------------------|
| **When schema is enforced** | At write time by the database | At read time by application code |
| **Error detection** | Immediate: bad data is rejected | Delayed: bad data is stored, fails at read |
| **Schema evolution** | ALTER TABLE (can be expensive) | Just start writing new fields |
| **Data quality** | Higher: database enforces constraints | Lower: application must validate |
| **Flexibility** | Lower: must define schema upfront | Higher: can iterate quickly |
| **Best for** | Structured data with known schema | Semi-structured data with evolving schema |

### Practical Guidance

Use schema-on-write when:
- Data integrity is critical (financial, medical, legal)
- Multiple applications share the same database
- You need complex queries across the dataset

Use schema-on-read when:
- The schema is evolving rapidly (early-stage product)
- Data comes from external sources with varying structure
- Each record type is accessed as a self-contained unit

---

## Query Languages Compared

### SQL (Relational)

Strengths: Mature, standardized, powerful optimizer, excellent tooling.
Weaknesses: Verbose for hierarchical data, recursive queries are awkward.

### MongoDB Query Language (Document)

```javascript
db.users.find({
  "addresses.state": "CA",
  "orders.created_at": { $gte: ISODate("2024-01-01") }
})
```

Strengths: Natural for document traversal, aggregation pipeline is powerful.
Weaknesses: No joins (until v3.2 $lookup, still limited), complex aggregations are hard to read.

### Cypher (Graph)

Strengths: Pattern matching for relationships, readable path expressions.
Weaknesses: Limited ecosystem, fewer tools and integrations.

### MapReduce (Batch)

```javascript
// Word count in MapReduce
map: function() { this.text.split(" ").forEach(w => emit(w, 1)); }
reduce: function(key, values) { return Array.sum(values); }
```

Strengths: Horizontally scalable, handles massive datasets.
Weaknesses: Low-level, hard to compose, high latency.

---

## Data Model Evolution

### Adding Fields

- **Relational:** `ALTER TABLE users ADD COLUMN phone VARCHAR(20);` -- all rows get NULL until updated
- **Document:** Just start including `phone` in new documents; old documents simply lack the field
- **Graph:** Add a new property to nodes; existing nodes are unaffected

### Changing Relationships

- **Relational:** Add a junction table for many-to-many; migrate data
- **Document:** Restructure documents; may require a migration script for existing data
- **Graph:** Add new edge types between existing nodes

### Breaking Changes

In all models, renaming or removing fields requires backward-compatible migration:

1. **Expand:** Add new field alongside old
2. **Migrate:** Backfill new field from old
3. **Contract:** Remove old field once all readers use new field

This expand-migrate-contract pattern works regardless of data model.

---

## Polyglot Persistence

Most real-world systems benefit from using multiple data stores, each chosen for its strengths:

| Use Case | Data Store | Reason |
|----------|-----------|--------|
| **Transactional records** | PostgreSQL | ACID, joins, referential integrity |
| **User sessions** | Redis | Sub-millisecond reads, TTL expiration |
| **Full-text search** | Elasticsearch | Inverted indexes, relevance scoring |
| **Activity feed** | Cassandra | High write throughput, time-series partitioning |
| **Recommendation graph** | Neo4j | Relationship traversal, path algorithms |
| **File/blob storage** | S3 | Unlimited capacity, durability |
| **Analytics** | ClickHouse/BigQuery | Column-oriented, fast aggregation |

### Polyglot Challenges

- **Data consistency:** How do you keep PostgreSQL and Elasticsearch in sync? Change data capture (CDC) is the standard answer
- **Operational complexity:** Each store requires monitoring, backup, and expertise
- **Query routing:** Application must know which store to query for which use case

### When to Stay Monoglot

Use a single database when:
- Your team is small and operational complexity is a bigger risk than suboptimal performance
- Your data fits comfortably in one model (most CRUD apps)
- Your query patterns are uniform (all point lookups, or all analytical scans)

Add a second store only when you have measured evidence that the current store cannot serve a specific access pattern.

---

## Data Model Migration Patterns

### Relational to Document

Common when an application starts with a relational database but finds that most queries fetch entire aggregate objects (user profiles, product listings) rather than joining across tables. The migration pattern:

1. Identify aggregates that are always fetched together
2. Denormalize related tables into nested document structures
3. Accept data duplication for fields that are shared across aggregates (e.g., category names stored in both the category table and embedded in product documents)
4. Maintain a relational database for data that genuinely requires joins and referential integrity

### Document to Relational

Common when an application starts with a document database but discovers increasing need for cross-document queries, reporting, or referential integrity. Warning signs include frequent application-level joins, growing inconsistency from denormalized data, and complex aggregation queries that fight the document model.

### Adding a Graph Layer

When relationships between entities become a first-class concept (recommendations, fraud detection, knowledge graphs), adding a graph database alongside existing stores is often more practical than migrating. The graph database handles traversal queries while the primary store handles CRUD operations. Data synchronization between the stores is typically handled through CDC or periodic ETL jobs.

# Reliability and Fault Tolerance

Reliability means the system continues to work correctly even when things go wrong. Things going wrong are called faults, and a system that can cope with faults is called fault-tolerant. The distinction between a fault and a failure is critical: a fault is when one component of the system deviates from its specification, while a failure is when the system as a whole stops providing the required service.


## Table of Contents
1. [Faults vs. Failures](#faults-vs-failures)
2. [Types of Faults](#types-of-faults)
3. [Reliability Metrics](#reliability-metrics)
4. [Detecting Faults in Distributed Systems](#detecting-faults-in-distributed-systems)
5. [Byzantine Faults](#byzantine-faults)
6. [Safety and Liveness](#safety-and-liveness)
7. [Designing for Reliability](#designing-for-reliability)
8. [Practical Reliability Patterns](#practical-reliability-patterns)

---

## Faults vs. Failures

| Term | Definition | Example |
|------|-----------|---------|
| **Fault** | One component deviating from its specification | A disk sector becomes unreadable |
| **Failure** | The system as a whole stops providing the required service | The entire website goes down |
| **Fault tolerance** | Designing the system so that faults don't become failures | RAID mirrors data across disks so one disk fault doesn't cause data loss |

The goal is not to prevent all faults (that is impossible) but to design systems that prevent faults from causing failures.

---

## Types of Faults

### Hardware Faults

Hardware faults are random and largely independent. The probability that two unrelated hardware components fail at the same time is very low.

| Component | Typical Failure Rate | Mitigation |
|-----------|---------------------|------------|
| **Hard disk** | MTTF ~10-50 years per drive | RAID, replicated storage |
| **RAM** | ~0.2% of DIMMs per year | ECC memory, replication |
| **Power supply** | Varies by quality | Dual power supplies, UPS, generators |
| **Network** | Partial failures common | Redundant paths, failover routing |
| **CPU** | Extremely rare | Multi-node redundancy |

**Key insight:** As cluster sizes grow, hardware faults become common events. With 10,000 disks (MTTF 10 years), expect roughly 3 disk failures per day. Systems must handle hardware faults as routine events, not exceptional emergencies.

### Software Faults

Software faults are systematic and correlated. A bug that crashes one node is likely to crash all nodes running the same software. Software faults are more dangerous than hardware faults because they are correlated -- they affect many nodes simultaneously.

**Common software faults:**
- A bug triggered by unusual input that crashes every instance processing that input
- A runaway process consuming all CPU, memory, or disk on every machine
- A cascading failure where one service's slowdown triggers timeouts in dependent services
- A leap second bug that affects every NTP-synchronized server simultaneously

**Mitigations:**
- **Process isolation:** Run services in separate processes or containers so a crash in one doesn't affect others
- **Input validation:** Reject malformed input at the boundary before it reaches core logic
- **Circuit breakers:** Detect when a dependency is failing and stop sending requests, preventing cascade
- **Chaos engineering:** Deliberately inject faults to discover weaknesses before they cause outages
- **Gradual rollouts:** Deploy new code to a small percentage of servers first; monitor before rolling out widely

### Human Errors

Humans are the leading cause of outages. Studies show that configuration errors cause the majority of production incidents -- not hardware or software failures.

**Mitigations:**
- **Design systems that minimize opportunity for error:** Well-designed APIs, admin interfaces, and configurations make it hard to do the wrong thing. Sensible defaults, validation, and dry-run modes.
- **Provide sandbox environments:** Allow engineers to experiment and test safely without affecting production.
- **Test at all levels:** Unit tests, integration tests, property-based tests, chaos tests. Automated testing catches errors that humans introduce.
- **Quick rollback:** Make it fast and easy to roll back a bad deployment. Feature flags allow disabling new code without redeploying.
- **Monitoring and alerting:** Detect problems early through metrics, dashboards, and alerts. If something goes wrong, you want to know in minutes, not hours.
- **Blameless postmortems:** Focus on systemic improvements, not individual blame. If a human error caused an outage, ask why the system allowed that error to cause an outage.

---

## Reliability Metrics

### Availability

Availability is the percentage of time the system is operational:

```
Availability = Uptime / (Uptime + Downtime)
```

| Availability | Downtime per Year | Downtime per Month |
|-------------|-------------------|--------------------|
| **99% (two nines)** | 3.65 days | 7.3 hours |
| **99.9% (three nines)** | 8.76 hours | 43.8 minutes |
| **99.99% (four nines)** | 52.6 minutes | 4.38 minutes |
| **99.999% (five nines)** | 5.26 minutes | 26.3 seconds |

**Key insight:** Each additional nine is roughly 10x harder to achieve. Going from 99.9% to 99.99% requires fundamentally different architecture, not just better operations.

### Durability

Durability is the probability that data, once written, will not be lost:

- **S3 Standard:** 99.999999999% (11 nines) durability -- designed to sustain the loss of data in two facilities simultaneously
- **Single-disk:** ~99.5% over 5 years (depending on disk failure rate)
- **RAID-1:** ~99.99% over 5 years
- **Replicated across 3 data centers:** Approaches 11+ nines

### Mean Time Between Failures (MTBF) and Mean Time To Recovery (MTTR)

```
Availability = MTBF / (MTBF + MTTR)
```

**Implication:** You can improve availability by either increasing MTBF (making failures less frequent) or decreasing MTTR (recovering faster). In practice, reducing MTTR is often more cost-effective because you can't eliminate all faults, but you can recover from them faster.

---

## Detecting Faults in Distributed Systems

### Timeouts

Timeouts are the primary mechanism for detecting faults in distributed systems. If a node doesn't respond within the timeout period, it is considered failed.

**The timeout dilemma:**
- **Too short:** False positives -- a slow but healthy node is declared dead, causing unnecessary failover, load redistribution, and potentially split-brain
- **Too long:** Slow detection -- a truly dead node continues to receive requests that fail, increasing latency and error rates for users

**Choosing timeouts:**
- Measure the p99 response time of healthy nodes
- Set the timeout to p99 * 2 or p99 + a fixed margin (e.g., 1 second)
- Use adaptive timeouts that adjust based on observed latency (Phi Accrual Failure Detector)

### Heartbeats

Nodes periodically send heartbeat messages to indicate they are alive. If a heartbeat is missed, the node may be considered failed.

```
Node A -> Heartbeat every 1 second -> Monitor
Node B -> Heartbeat every 1 second -> Monitor

If Monitor receives no heartbeat from Node B for 3 seconds:
  -> Declare Node B potentially failed
  -> Trigger health check or failover
```

**Heartbeat patterns:**
- **Push-based:** Each node sends heartbeats to a central monitor or to other nodes
- **Pull-based:** A monitor periodically polls each node for status
- **Gossip-based:** Each node gossips its status to random peers; failure information spreads epidemically

### Failure Detectors

A failure detector is an abstraction that encapsulates the logic of deciding whether a node is alive or dead. Properties of failure detectors:

| Property | Meaning |
|----------|---------|
| **Completeness** | Every failed node is eventually detected |
| **Accuracy** | No healthy node is incorrectly declared failed |

In asynchronous networks, no failure detector can guarantee both properties simultaneously. Practical failure detectors sacrifice accuracy (may occasionally declare healthy nodes as failed) to ensure completeness (never miss a truly failed node).

---

## Byzantine Faults

### What Are Byzantine Faults?

A Byzantine fault occurs when a node behaves in an arbitrary and potentially malicious way: sending conflicting information to different peers, lying about its state, or corrupting data intentionally.

### When Byzantine Fault Tolerance Matters

| Context | Needed? | Why |
|---------|---------|-----|
| **Internal datacenter** | No | You trust your own servers; if one is compromised, you have bigger problems |
| **Public blockchain** | Yes | Participants are mutually untrusting; any node may be malicious |
| **Aerospace/nuclear systems** | Sometimes | Radiation can flip bits, causing non-crash arbitrary behavior |
| **Multi-organization systems** | Sometimes | If organizations don't trust each other, Byzantine tolerance may be needed |

### Why Most Systems Ignore Byzantine Faults

Byzantine fault tolerance requires 3f + 1 nodes to tolerate f Byzantine faults. This means tolerating 1 malicious node requires 4 nodes, and tolerating 2 requires 7. The overhead is substantial.

Most systems instead assume a crash-stop or crash-recovery model:
- **Crash-stop:** A faulty node simply stops and never comes back
- **Crash-recovery:** A faulty node stops but may come back with its state intact (from durable storage)

These simpler fault models are sufficient for the vast majority of data systems.

---

## Safety and Liveness

### Definitions

| Property | Definition | Example |
|----------|-----------|---------|
| **Safety** | Nothing bad happens | No two nodes are elected leader simultaneously (no split-brain) |
| **Liveness** | Something good eventually happens | A failed node is eventually detected; a client request eventually receives a response |

### Why the Distinction Matters

In distributed systems, you can always guarantee safety properties, but liveness properties may be temporarily violated:

- **Safety must always hold.** If a safety property is violated even once, the violation is irrecoverable. You can point to a specific moment when the property was violated.
- **Liveness may be temporarily violated.** A liveness violation means something hasn't happened yet, but it may happen in the future. You can't point to a specific moment when it was violated.

**Example: Leader election**
- Safety: At most one leader at any time (must always hold)
- Liveness: A leader is eventually elected (may be temporarily violated during an election)

### Practical Implications

When designing distributed systems, prioritize safety over liveness:
- It is better for the system to be temporarily unavailable (liveness violation) than to produce incorrect results (safety violation)
- A consensus algorithm that never elects a leader is safe (no split-brain) but useless (no liveness)
- The art is achieving both safety and liveness under realistic assumptions about network and node behavior

---

## Designing for Reliability

### Defense in Depth

No single mechanism provides complete reliability. Layer multiple defenses:

```
Layer 1: Input validation and sanitization
Layer 2: Application-level error handling and retries
Layer 3: Database transactions and constraints
Layer 4: Replication and failover
Layer 5: Backups and disaster recovery
Layer 6: Monitoring, alerting, and incident response
```

### Failure Mode Analysis

For each component, ask:
1. **How can it fail?** (crash, slow down, return wrong answer, become unreachable)
2. **What happens when it fails?** (impact on dependent components and users)
3. **How will we detect the failure?** (monitoring, health checks, alerts)
4. **How will we recover?** (automatic failover, manual intervention, restore from backup)
5. **How do we prevent it?** (redundancy, testing, capacity planning)

### Chaos Engineering Principles

Chaos engineering proactively injects faults to discover weaknesses:

| Experiment | What It Tests | Tools |
|-----------|--------------|-------|
| **Kill a node** | Failover and recovery | Chaos Monkey, kill -9 |
| **Network partition** | Partition tolerance, split-brain prevention | tc (traffic control), iptables, Toxiproxy |
| **Clock skew** | Time-dependent logic, lease expiration | libfaketime, NTP manipulation |
| **Disk full** | Logging, WAL, temporary files | dd, fallocate |
| **Slow responses** | Timeout handling, circuit breakers | Toxiproxy, tc netem |
| **DNS failure** | Service discovery fallback | iptables blocking port 53 |
| **Certificate expiration** | TLS handling, renewal processes | Short-lived test certificates |

### The Recovery-Oriented Computing Approach

Instead of trying to prevent all failures, optimize for fast recovery:

1. **Micro-reboots:** Restart individual components instead of entire systems
2. **Undo support:** Every action has an undo; rollback is always available
3. **Redundancy at every level:** No single point of failure from hardware to application
4. **Monitoring is a first-class feature:** Not an afterthought; built into every component from day one
5. **Automation over documentation:** Runbooks become scripts; manual procedures become automated workflows

---

## Practical Reliability Patterns

### Retry with Exponential Backoff and Jitter

```python
def retry_with_backoff(operation, max_retries=3, base_delay=1.0):
    for attempt in range(max_retries):
        try:
            return operation()
        except TransientError:
            if attempt == max_retries - 1:
                raise
            delay = base_delay * (2 ** attempt)
            jitter = random.uniform(0, delay * 0.5)
            time.sleep(delay + jitter)
```

Jitter prevents thundering herd: if 1000 clients all retry at exactly the same time, they overload the recovering service.

### Circuit Breaker

```
States: CLOSED -> OPEN -> HALF-OPEN -> CLOSED

CLOSED: Requests pass through normally
  -> If failure rate exceeds threshold: transition to OPEN

OPEN: All requests immediately fail (fast fail, no network call)
  -> After timeout period: transition to HALF-OPEN

HALF-OPEN: Allow a small number of test requests
  -> If test requests succeed: transition to CLOSED
  -> If test requests fail: transition back to OPEN
```

### Bulkhead Pattern

Isolate components so that a failure in one doesn't exhaust shared resources:

```
Thread Pool A (20 threads): Service A calls
Thread Pool B (20 threads): Service B calls
Thread Pool C (10 threads): Service C calls

If Service B becomes slow and exhausts its 20 threads,
Services A and C are unaffected -- they have their own pools.
```

### Health Check Endpoints

Every service should expose a health check endpoint that reports:

```json
{
  "status": "healthy",
  "checks": {
    "database": {"status": "healthy", "latency_ms": 5},
    "redis": {"status": "healthy", "latency_ms": 1},
    "disk": {"status": "healthy", "free_gb": 42},
    "memory": {"status": "healthy", "used_percent": 65}
  },
  "version": "2.3.1",
  "uptime_seconds": 86400
}
```

Load balancers and orchestrators (Kubernetes) use health checks to route traffic away from unhealthy instances and restart failing ones.

# Partitioning

Partitioning (also called sharding) divides a large dataset into smaller subsets called partitions, each stored on a different node. The goal is to spread data and query load evenly across machines, enabling horizontal scaling beyond the capacity of a single node.

## Why Partition?

A single database node has hard limits:
- **Storage capacity:** A single disk or SSD has a maximum size
- **Write throughput:** A single CPU can process a limited number of writes per second
- **Read throughput:** Even with caching, a single node can serve a limited number of concurrent reads

Partitioning breaks through all three limits by distributing data across multiple nodes. Each node handles a fraction of the total workload.

---

## Key-Range Partitioning

### How It Works

Assign a continuous range of keys to each partition, similar to volumes of an encyclopedia:

```
Partition 1: keys A-E
Partition 2: keys F-J
Partition 3: keys K-O
Partition 4: keys P-T
Partition 5: keys U-Z
```

The ranges are not necessarily evenly spaced -- they are chosen to distribute data evenly based on the actual key distribution.

### Strengths

- **Efficient range queries:** All keys in a range are on the same partition, so range scans are local and fast
- **Natural ordering:** Data is stored in sorted order within each partition, supporting ORDER BY queries
- **Good for time-series:** Partitioning by time range keeps recent data together

### Weaknesses

- **Hotspots on sequential keys:** If the partition key is a timestamp, all writes go to the partition for the current time period, creating a write hotspot
- **Uneven distribution:** Key ranges that looked balanced at partition creation may become skewed as data grows
- **Manual or complex rebalancing:** Range boundaries may need adjustment as data distribution changes

### Avoiding Time-Series Hotspots

Instead of partitioning by timestamp alone, use a composite key:

```
Partition key: (sensor_id, date)

Sensor 1, 2024-01-01 -> Partition A
Sensor 2, 2024-01-01 -> Partition B
Sensor 1, 2024-01-02 -> Partition A
Sensor 3, 2024-01-01 -> Partition C
```

This distributes writes across partitions (different sensors go to different partitions) while preserving the ability to scan one sensor's data in time order.

### Databases Using Key-Range Partitioning

- HBase (row key ranges)
- Bigtable (row key ranges)
- MongoDB (range-based sharding option)

---

## Hash Partitioning

### How It Works

Apply a hash function to the partition key and assign hash ranges to partitions:

```
hash(key) mod N = partition number

hash("user_123") = 0x7A3F... -> Partition 3
hash("user_456") = 0x1B2C... -> Partition 1
hash("user_789") = 0xE4D1... -> Partition 5
```

### Consistent Hashing

Standard `hash mod N` is problematic when adding or removing nodes because it reassigns most keys. Consistent hashing solves this by mapping both keys and nodes onto a ring:

```
Ring positions: 0 ... 2^32

Nodes:  A at position 1000, B at position 5000, C at position 9000
Key:    hash("user_123") = 3500 -> assigned to Node B (next node clockwise)
```

When a node is added or removed, only the keys between adjacent nodes are reassigned, minimizing data movement.

### Virtual Nodes (Vnodes)

Each physical node is assigned multiple positions (virtual nodes) on the ring, typically 256 per node. This:
- Distributes data more evenly (random positions may cluster otherwise)
- Enables proportional assignment (a more powerful node gets more virtual nodes)
- Smooths rebalancing (adding a node moves small chunks from many existing nodes)

### Strengths

- **Even distribution:** Hash functions distribute keys uniformly, avoiding hotspots from key distribution skew
- **Simple assignment:** Given the key, you can compute the partition without a lookup table

### Weaknesses

- **No range queries:** Hash destroys sort order, so range scans require querying all partitions (scatter-gather)
- **Hot keys still possible:** If one key receives disproportionate traffic (e.g., a celebrity's user ID), hashing doesn't help

### Databases Using Hash Partitioning

- Cassandra (default partitioner: Murmur3)
- DynamoDB (hash of partition key)
- Riak (consistent hashing)
- MongoDB (hash-based sharding option)

---

## Secondary Index Partitioning

When you need to query data by something other than the partition key, you need secondary indexes. There are two approaches to partitioning secondary indexes.

### Local Secondary Indexes (Document-Partitioned)

Each partition maintains its own secondary index covering only the data in that partition:

```
Partition 1: primary data A-M, local index on "color"
Partition 2: primary data N-Z, local index on "color"

Query: SELECT * WHERE color = 'red'
  -> Must query BOTH partitions (scatter-gather)
  -> Each checks its local index
  -> Results are merged
```

**Strengths:**
- Writes are local: updating the secondary index only affects one partition
- Simple to maintain: each partition is self-contained

**Weaknesses:**
- Reads require scatter-gather across all partitions
- Latency is determined by the slowest partition (tail latency)

**Used by:** MongoDB, Cassandra, Elasticsearch, SolrCloud

### Global Secondary Indexes (Term-Partitioned)

The secondary index is itself partitioned, but independently of the primary data:

```
Primary data: partitioned by user_id
Global index on "color": partitioned by color value
  Index partition 1: color A-M (all reds across all primary partitions)
  Index partition 2: color N-Z

Query: SELECT * WHERE color = 'red'
  -> Query index partition 1 only
  -> Get list of document IDs
  -> Fetch documents from their primary partitions
```

**Strengths:**
- Reads are efficient: query only the relevant index partition
- No scatter-gather for indexed queries

**Weaknesses:**
- Writes require updating a remote partition (cross-partition write)
- Index updates are often asynchronous, meaning the index may be stale
- More complex distributed transaction requirements

**Used by:** DynamoDB (global secondary indexes), Amazon Aurora

---

## Rebalancing Strategies

As data grows or nodes are added/removed, partitions must be rebalanced.

### Strategy 1: Fixed Number of Partitions

Create many more partitions than nodes (e.g., 1000 partitions for 10 nodes). Each node hosts multiple partitions. When a node is added, some partitions move from existing nodes to the new node.

```
Before: 3 nodes, 12 partitions (4 per node)
After adding node 4: 4 nodes, 12 partitions (3 per node)
Move 1 partition from each existing node to the new node
```

**Strengths:** Simple, no re-partitioning needed, proportional load balancing
**Weaknesses:** Must choose partition count upfront; too few means large partitions, too many means overhead

**Used by:** Elasticsearch, Riak, Couchbase, Voldemort

### Strategy 2: Dynamic Partitioning

Start with one partition. When a partition grows beyond a threshold (e.g., 10GB), split it in half. When it shrinks below a threshold, merge it with a neighbor.

**Strengths:** Adapts to data size automatically; no upfront sizing decisions
**Weaknesses:** Single partition at start means single-node bottleneck until first split; can cause split storms under rapid growth

**Used by:** HBase, RethinkDB, MongoDB (with key-range sharding)

### Strategy 3: Proportional to Nodes

Keep a fixed number of partitions per node. When a node is added, it splits some existing partitions; when removed, its partitions are merged into others.

**Strengths:** Partition count grows with cluster size; each partition stays manageable
**Weaknesses:** Splitting introduces brief unavailability for the affected partition

**Used by:** Cassandra (with vnodes)

---

## Request Routing

How does a client know which node holds the partition for a given key?

### Approach 1: Client-Side Routing

The client knows the partition assignment and connects directly to the correct node:

```
Client: hash("user_123") -> Partition 3 -> Node B
Client connects directly to Node B
```

Requires the client to maintain a copy of the partition map. Used by Cassandra drivers.

### Approach 2: Routing Tier (Proxy)

A separate routing tier receives all requests and forwards them to the correct node:

```
Client -> Proxy -> determines partition -> forwards to correct Node
```

Used by: MongoDB (mongos router), Twemproxy (for Redis/Memcached)

### Approach 3: Any-Node Contact

Client contacts any node; that node forwards the request if it doesn't own the partition:

```
Client -> Node A -> "Not my partition" -> forwards to Node B
```

Used by: Cassandra (coordinator pattern), CockroachDB

### Service Discovery

All approaches need to know the current partition-to-node mapping. Options:

- **ZooKeeper/etcd:** Centralized configuration service that tracks which partitions are on which nodes; nodes register themselves; routing layer watches for changes
- **Gossip protocol:** Nodes gossip partition assignments to each other; eventually consistent but no central point of failure
- **DNS-based:** Simple but slow to update; suitable only for coarse-grained routing

---

## Handling Hotspots

### Why Hotspots Occur

Even with perfect hash distribution, application-level access patterns create hotspots:

- **Celebrity problem:** A single user or entity receives vastly more traffic than others
- **Temporal hotspots:** Events cause sudden spikes on specific keys (product launch, breaking news)
- **Sequential keys:** Auto-incrementing IDs or timestamps concentrate writes

### Mitigation Strategies

| Strategy | How It Works | Trade-off |
|----------|-------------|-----------|
| **Key splitting** | Append random suffix (0-9) to hot keys; read from all 10 sub-keys and merge | 10x fan-out on reads; application complexity |
| **Write buffering** | Buffer writes to hot keys in memory; flush periodically | Eventual consistency; risk of data loss if buffer crashes |
| **Caching layer** | Cache hot reads in Redis/Memcached in front of the database | Stale data; cache invalidation complexity |
| **Rate limiting** | Limit requests to hot keys per client | Degrades user experience for hot content |
| **Application-level sharding** | Route hot entities to dedicated, scaled infrastructure | Operational complexity; special-case architecture |

### Detecting Hotspots

Monitor per-partition metrics:
- **Request rate per partition:** Compare against average; alert on 10x deviation
- **Latency per partition:** Hot partitions show higher p99 latency
- **CPU/IO utilization per node:** Uneven utilization signals partition skew
- **Key-level access counting:** Sample or log the most-accessed keys (most databases provide slow query logs or key-access statistics)

### Automatic Hotspot Detection

Some systems detect and mitigate hotspots automatically:
- **DynamoDB Adaptive Capacity:** Automatically isolates hot partitions onto dedicated throughput
- **Spanner:** Splits hot partitions when load exceeds threshold
- **CockroachDB:** Automatic range splitting and lease rebalancing based on load

# Replication

Replication means keeping a copy of the same data on multiple machines connected via a network. The reasons for replication are: keeping data geographically close to users (reduce latency), allowing the system to continue working even if some machines fail (increase availability), and scaling out the number of machines that can serve read queries (increase read throughput).

## Single-Leader Replication

### How It Works

One node is designated the leader (primary, master). All writes go to the leader, which writes data to its local storage and sends the change to all followers (replicas, secondaries) via a replication log. Followers apply the changes in the same order.

```
Client Writes --> Leader --> Replication Log --> Follower 1
                                            --> Follower 2
                                            --> Follower 3

Client Reads  --> Leader OR any Follower
```

### Synchronous vs. Asynchronous Replication

| Mode | Behavior | Trade-off |
|------|----------|-----------|
| **Synchronous** | Leader waits for follower confirmation before acknowledging write to client | Guaranteed durability on follower; higher write latency; follower outage blocks writes |
| **Asynchronous** | Leader acknowledges write immediately after local write; replicates in background | Lower write latency; leader failure can lose confirmed writes; follower may be stale |
| **Semi-synchronous** | One follower is synchronous, rest are asynchronous | Guarantees data exists on at least two nodes; practical compromise |

PostgreSQL, MySQL, and MongoDB all default to asynchronous replication. This means a write that the leader confirms can be lost if the leader crashes before replicating it.

### Leader Failover

When the leader fails, a follower must be promoted:

1. **Detect failure:** Typically via heartbeat timeouts (e.g., no response for 30 seconds)
2. **Choose new leader:** Consensus among remaining nodes, or human intervention
3. **Reconfigure system:** Clients must send writes to the new leader; old leader must become a follower when it recovers

**Failover dangers:**
- **Split-brain:** Two nodes both believe they are the leader; both accept writes, causing data divergence
- **Lost writes:** If the old leader had unreplicated writes, they are lost (or conflict with new leader's writes)
- **Stale routing:** Clients with cached leader addresses continue writing to the old leader

---

## Replication Lag Problems

Asynchronous replication introduces a delay between a write on the leader and its appearance on followers. This lag causes several anomalies.

### Read-Your-Own-Writes Violation

**Problem:** A user writes data (to the leader), then immediately reads it (from a follower that hasn't received the write yet). The user sees stale data and thinks their write was lost.

**Solutions:**
- Read from the leader for data the user has recently modified (e.g., always read your own profile from the leader)
- Track the timestamp of the user's last write; only read from followers that are caught up to that timestamp
- Client remembers the position in the replication log of its last write and waits for the follower to reach that position

### Monotonic Reads Violation

**Problem:** A user makes two reads in sequence and sees time go backward -- the second read returns older data than the first because it hit a different, more-lagged follower.

**Solutions:**
- Pin each user to a specific follower (session affinity) so consecutive reads go to the same replica
- Track the most recent read timestamp and ensure subsequent reads go to followers at least that current

### Consistent Prefix Reads Violation

**Problem:** Causally related writes appear out of order. A database stores a question and its answer; a reader sees the answer before the question because different partitions have different replication lag.

**Solutions:**
- Ensure causally related writes go to the same partition
- Use causal consistency tracking (vector clocks or Lamport timestamps)

---

## Multi-Leader Replication

### Use Cases

Multi-leader (active-active) replication allows writes at multiple data centers:

- **Multi-datacenter operation:** Each data center has its own leader; writes are fast locally and replicated asynchronously to other data centers
- **Offline-capable clients:** A mobile app with a local database acts as a leader; syncs with server when online (CouchDB, PouchDB model)
- **Collaborative editing:** Each user's local state acts as a leader; changes are merged asynchronously

### Conflict Resolution

When two leaders accept conflicting writes to the same record, the conflict must be resolved:

| Strategy | How It Works | Trade-off |
|----------|-------------|-----------|
| **Last writer wins (LWW)** | Assign a timestamp to each write; highest timestamp wins | Simple but discards concurrent writes; data loss is possible |
| **Merge values** | Combine conflicting values (e.g., union of sets) | Preserves data but only works for certain data types |
| **Application-level resolution** | Store all conflicting versions; let application code resolve | Most flexible but pushes complexity to the application |
| **CRDTs** | Use data structures mathematically guaranteed to converge | Automatic convergence; limited to specific data types (counters, sets, registers) |

### Conflict Example

```
Leader A: UPDATE users SET name = 'Alice Chen' WHERE id = 42;
Leader B: UPDATE users SET name = 'Alice Wang' WHERE id = 42;

LWW result: One name wins (other is silently lost)
Merge result: Not meaningful for names
CRDT (LWW-Register): Last timestamp wins, but the conflict is detected
Application resolution: Show user both versions, ask which is correct
```

---

## Leaderless Replication

### How It Works

In leaderless replication (used by Dynamo, Cassandra, Riak, Voldemort), there is no leader. Clients send writes to multiple replicas directly (or via a coordinator node). Reads also query multiple replicas and reconcile differences.

### Quorum Reads and Writes

Given `n` replicas, a write succeeds if acknowledged by `w` replicas, and a read succeeds if it reads from `r` replicas. As long as `w + r > n`, at least one of the read replicas will have the latest write.

**Common configurations:**
- `n=3, w=2, r=2`: Tolerates 1 unavailable node for both reads and writes
- `n=3, w=3, r=1`: Fastest reads (only need one node), but writes require all nodes
- `n=3, w=1, r=3`: Fastest writes, but reads must query all nodes
- `n=5, w=3, r=3`: Tolerates 2 unavailable nodes

### Sloppy Quorums and Hinted Handoff

When a node is unavailable, a strict quorum would reject the write. A sloppy quorum instead writes to a different node temporarily. When the original node recovers, the temporary node forwards the data (hinted handoff).

This improves write availability but weakens consistency guarantees -- `w + r > n` no longer guarantees reading the latest write because the writes may be on nodes outside the usual `n`.

### Read Repair and Anti-Entropy

Stale replicas need to be updated:

- **Read repair:** When a read query detects a stale replica (by comparing versions), the client writes the latest value back to the stale replica
- **Anti-entropy process:** A background process continuously compares data between replicas and copies missing data. Unlike replication logs, this does not preserve ordering.

---

## CRDTs: Conflict-Free Replicated Data Types

### What CRDTs Are

CRDTs are data structures that can be replicated across multiple nodes, where replicas can be updated independently and concurrently without coordination, and which mathematically guarantee eventual convergence.

### Common CRDT Types

| CRDT | Use Case | How It Works |
|------|----------|-------------|
| **G-Counter** | Counting (only increment) | Each node maintains its own counter; total = sum of all node counters |
| **PN-Counter** | Counting (increment and decrement) | Two G-Counters: one for increments, one for decrements; value = P - N |
| **G-Set** | Set (only add) | Union of all elements across all replicas |
| **OR-Set** | Set (add and remove) | Each element has unique tags; remove deletes specific tags |
| **LWW-Register** | Single value | Last write (by timestamp) wins |
| **MV-Register** | Single value | Keeps all concurrent versions; application resolves |

### CRDT Example: Collaborative Counter

```
Node A starts: count = 0
Node B starts: count = 0

Node A: increment -> local count_A = 1
Node B: increment -> local count_B = 1
Node B: increment -> local count_B = 2

After sync:
Both nodes: total = count_A + count_B = 1 + 2 = 3
```

No conflicts, no coordination, mathematically correct.

### CRDT Limitations

- Only work for specific data types (not arbitrary business logic)
- Can grow in memory over time (tombstones, version vectors)
- Eventual consistency only -- no guarantee of when convergence happens
- Complex to implement correctly from scratch; use libraries (Automerge, Yjs, delta-state CRDTs)

---

## Replication Topology Patterns

### Single-Leader Topologies

```
Star:     Leader --> F1, F2, F3, F4    (all followers connect to leader)
Chain:    Leader --> F1 --> F2 --> F3   (each follower replicates to next)
```

Star is simpler; chain reduces load on the leader but increases replication delay.

### Multi-Leader Topologies

```
All-to-All:   A <--> B <--> C, A <--> C    (every leader replicates to every other)
Star:          A <--> B, A <--> C            (one central leader relays)
Circular:      A --> B --> C --> A            (each sends to the next in a ring)
```

All-to-all is most fault-tolerant but creates more replication traffic. Circular and star topologies have single points of failure.

---

## Practical Replication Configurations

### PostgreSQL Streaming Replication

```
Primary (leader) --> Streaming Replication --> Standby 1 (sync)
                                           --> Standby 2 (async)
                                           --> Standby 3 (async)
```

- Write to primary; read from standbys for scaling
- Synchronous standby guarantees zero data loss on failover
- Asynchronous standbys may lag behind by seconds

### Cassandra (Leaderless)

```
Client --> Coordinator Node --> Replica 1 (write)
                            --> Replica 2 (write)
                            --> Replica 3 (write)

Read: Query 2 of 3 replicas (QUORUM), return most recent
```

- Replication factor (n) set per keyspace
- Consistency level (w, r) set per query
- Tunable consistency: `ONE` for speed, `QUORUM` for safety, `ALL` for strong consistency

### Redis Sentinel (Single-Leader with Automatic Failover)

```
Master --> Replica 1
       --> Replica 2

Sentinel 1, Sentinel 2, Sentinel 3 (monitor master, vote on failover)
```

- Sentinels detect master failure via heartbeats
- Majority of sentinels must agree before promoting a replica
- Client library queries sentinel for current master address

# Storage Engines

A storage engine is the component of a database that handles how data is written to and read from disk (or memory). Understanding storage engine internals is essential for predicting performance, choosing appropriate indexes, and avoiding pathological workloads.

## Two Families of Storage Engines

All storage engines face a fundamental trade-off: optimizing for write performance or read performance. The two dominant approaches are log-structured engines (optimized for writes) and page-oriented engines (balanced reads and writes).

---

## Log-Structured Engines: LSM Trees and SSTables

### How LSM Trees Work

LSM (Log-Structured Merge) trees use a multi-level structure:

1. **Memtable:** An in-memory balanced tree (typically a red-black tree or skip list) that receives all writes
2. **Flush to SSTable:** When the memtable reaches a size threshold (typically 64MB-256MB), it is written to disk as a Sorted String Table (SSTable) -- a file of key-value pairs sorted by key
3. **Compaction:** Background processes merge multiple SSTables into fewer, larger SSTables, removing deleted keys and resolving duplicates

### Write Path

```
Client Write
    |
    v
Write-Ahead Log (WAL) -- sequential append for durability
    |
    v
Memtable (in-memory sorted structure)
    |
    v  (when full)
SSTable on disk (sorted, immutable file)
    |
    v  (background)
Compaction: merge SSTables into larger, consolidated files
```

### Read Path

```
Client Read
    |
    v
Check Memtable (most recent writes)
    |  (miss)
    v
Check Bloom Filters for each SSTable level
    |  (possible match)
    v
Binary search within SSTable
    |
    v
Return value (or not found)
```

### Compaction Strategies

| Strategy | How It Works | Trade-off |
|----------|-------------|-----------|
| **Size-tiered** | SSTables of similar size are merged together | Better write throughput; more space amplification |
| **Leveled** | SSTables are organized into levels of increasing size; each level is non-overlapping | Better read performance and space efficiency; higher write amplification |

- **Cassandra** defaults to size-tiered compaction
- **RocksDB** and **LevelDB** use leveled compaction
- Choice depends on read/write ratio and disk space constraints

### LSM Tree Strengths

- **Sequential writes:** All disk writes are sequential appends, which is much faster than random I/O on both HDDs and SSDs
- **High write throughput:** Buffering in memory and batch-flushing to disk minimizes disk operations per write
- **Compression:** Sorted SSTables compress well, reducing storage costs
- **No fragmentation:** Compaction produces fresh, defragmented files

### LSM Tree Weaknesses

- **Read amplification:** A point read may need to check the memtable plus multiple SSTables at different levels
- **Write amplification:** A single logical write may be written and rewritten multiple times through compaction (typical: 10-30x)
- **Compaction interference:** Background compaction consumes CPU and I/O bandwidth, causing latency spikes if not tuned
- **Space amplification:** During compaction, both old and new SSTables exist temporarily, requiring 2x space

### Databases Using LSM Trees

- Cassandra, ScyllaDB, HBase (size-tiered)
- RocksDB, LevelDB (leveled)
- CockroachDB, TiKV (RocksDB-based)

---

## Page-Oriented Engines: B-Trees

### How B-Trees Work

B-trees organize data in fixed-size pages (typically 4KB-16KB) arranged in a balanced tree structure:

1. **Root page:** Contains keys and pointers to child pages
2. **Internal pages:** Contains keys that split the key space and pointers to child pages
3. **Leaf pages:** Contains the actual key-value pairs (or pointers to heap file rows)

A B-tree with a branching factor of 500 and 4 levels can store up to 256TB of data (500^4 pages).

### Write Path

```
Client Write
    |
    v
Write-Ahead Log (WAL) -- for crash recovery
    |
    v
Traverse B-tree from root to leaf
    |
    v
Update the leaf page in place
    |  (if page is full)
    v
Split page: create two half-full pages, update parent pointer
```

### Read Path

```
Client Read
    |
    v
Start at root page
    |
    v
Binary search within page for correct child pointer
    |
    v
Follow pointer to next level
    |  (repeat log(n) times)
    v
Reach leaf page, binary search for key
    |
    v
Return value
```

### B-Tree Strengths

- **Predictable read latency:** Every lookup follows the same number of page accesses (tree depth), typically 3-4 for practical databases
- **Efficient point lookups:** O(log n) with small constants due to high branching factor
- **Mature and battle-tested:** 40+ years of optimization, well-understood behavior
- **Good range scan performance:** Leaf pages are often linked, allowing sequential scanning

### B-Tree Weaknesses

- **Write amplification:** Even a small update requires rewriting an entire page (typically 4KB-16KB)
- **Page splits:** When a page is full, it splits into two, requiring parent page updates (can cascade)
- **Fragmentation:** Over time, pages become partially full, wasting space
- **Concurrency control:** In-place updates require careful locking (latches) to prevent torn reads

### Databases Using B-Trees

- PostgreSQL, MySQL/InnoDB, Oracle, SQL Server
- SQLite
- Most traditional relational databases

---

## LSM Trees vs. B-Trees: Decision Guide

| Factor | LSM Trees | B-Trees |
|--------|-----------|---------|
| **Write throughput** | Higher (sequential writes) | Lower (random in-place updates) |
| **Read latency** | Less predictable (multiple levels) | More predictable (fixed tree depth) |
| **Space efficiency** | Better (compaction removes dead entries) | Worse (fragmentation, partial pages) |
| **Write amplification** | Higher (compaction rewrites) | Lower per write, but each write is a full page |
| **Compression** | Better (sorted data compresses well) | Moderate |
| **Concurrency** | No in-place updates, simpler | Requires page-level latching |
| **Maturity** | Newer, less predictable edge cases | Decades of production hardening |
| **Best for** | Write-heavy, append-heavy workloads | Mixed read/write OLTP |

### Rules of Thumb

- **Write-heavy with few reads:** LSM tree (Cassandra, RocksDB)
- **Read-heavy with indexed lookups:** B-tree (PostgreSQL, MySQL)
- **Mixed OLTP:** B-tree, unless write throughput is the bottleneck
- **Time-series ingestion:** LSM tree (high sequential write rate)
- **When in doubt:** Start with B-tree (PostgreSQL); it handles most workloads well

---

## Column-Oriented Storage

### The Problem with Row Storage for Analytics

Analytical queries typically access a few columns across millions or billions of rows:

```sql
SELECT product_category, SUM(revenue), COUNT(*)
FROM sales
WHERE sale_date BETWEEN '2024-01-01' AND '2024-12-31'
GROUP BY product_category;
```

In a row-oriented store, this query reads entire rows (all columns) even though it only needs three columns. With 100 columns and 1 billion rows, you read 100x more data than necessary.

### How Column Storage Works

Column-oriented storage stores each column separately:

```
Row store:          Column store:
[id, name, age]     ids:   [1, 2, 3, ...]
[1, Alice, 30]      names: [Alice, Bob, Carol, ...]
[2, Bob, 25]        ages:  [30, 25, 28, ...]
[3, Carol, 28]
```

### Column Storage Benefits

- **I/O reduction:** Only read the columns your query needs
- **Compression:** Values in a single column have similar data types and distributions, enabling excellent compression (often 10:1)
- **Vectorized processing:** Modern CPUs process arrays of same-typed values much faster than individual rows (SIMD instructions)
- **Bitmap indexes:** Column values can be efficiently indexed with bitmaps for fast filtering

### Column Storage Implementations

| Database | Type | Key Feature |
|----------|------|-------------|
| **ClickHouse** | Column OLAP | MergeTree engine, real-time aggregation |
| **Apache Parquet** | File format | Columnar storage for Hadoop/Spark/data lakes |
| **Apache ORC** | File format | Optimized for Hive, predicate pushdown |
| **BigQuery** | Cloud OLAP | Serverless, automatic optimization |
| **Redshift** | Cloud OLAP | Zone maps, sort keys for pruning |
| **DuckDB** | Embedded OLAP | In-process, Parquet-native |

---

## In-Memory Databases

### Why In-Memory is Fast

The performance advantage of in-memory databases is not simply "RAM is faster than disk." RAM-based systems are fast because they avoid the overhead of encoding data into disk-friendly formats. In-memory data structures (hash tables, skip lists, trees) can be used directly without serialization.

### In-Memory Database Types

| Database | Model | Persistence | Use Case |
|----------|-------|-------------|----------|
| **Redis** | Key-value, data structures | Optional (RDB snapshots, AOF log) | Caching, sessions, rate limiting, queues |
| **Memcached** | Key-value | None | Simple caching |
| **VoltDB** | Relational | Durable (WAL + replication) | High-throughput OLTP with serializability |
| **SAP HANA** | Relational + column | Durable | Mixed OLTP/OLAP |

### Anti-Heap: Persistence for In-Memory Databases

In-memory databases that need durability use several techniques:

- **Write-ahead log (WAL):** Append every write to disk sequentially; on crash, replay the log
- **Periodic snapshots:** Write the entire in-memory state to disk at intervals
- **Replication:** Keep copies on other machines; if one crashes, another has the data
- **Battery-backed RAM:** Hardware guarantee that RAM contents survive power loss

The WAL approach means writes are actually written to disk, but reads never touch disk. The disk serves only as a durability mechanism, not as a primary data structure.

---

## Choosing a Storage Engine: Decision Framework

### Step 1: Classify Your Workload

| Question | Answer Determines |
|----------|-------------------|
| What is your read:write ratio? | LSM (write-heavy) vs. B-tree (balanced/read-heavy) |
| Do you need point lookups or range scans? | Hash index (point) vs. B-tree/LSM (range) |
| Is your data mostly queried by row or by column? | Row store (OLTP) vs. column store (OLAP) |
| Does your data fit in memory? | In-memory store for sub-millisecond latency |
| What latency percentile matters (p50 vs. p99)? | B-tree for predictable p99; LSM for better p50 |

### Step 2: Consider Operational Factors

- **Team expertise:** Use what your team knows unless there is a compelling reason to switch
- **Ecosystem:** Consider drivers, ORMs, monitoring tools, and backup solutions
- **Managed services:** Cloud-managed databases reduce operational burden significantly
- **Vendor lock-in:** Open-source engines provide more flexibility

### Step 3: Test With Your Actual Workload

Benchmarks from the internet are misleading. They test different hardware, different data sizes, different access patterns, and different configurations. The only benchmark that matters is one that uses your data, your queries, and your expected concurrency.

Tools for benchmarking:
- **YCSB (Yahoo Cloud Serving Benchmark):** Standard workload generator for key-value stores
- **TPC-C:** Standard OLTP benchmark
- **TPC-H:** Standard OLAP benchmark
- **pgbench:** PostgreSQL-specific benchmark
- **sysbench:** MySQL-specific benchmark

# Transactions and Consistency

Transactions are an abstraction layer that simplifies the programming model for applications accessing a database. They bundle multiple reads and writes into a single logical operation that either succeeds completely (commit) or fails completely (abort), with no partial results visible to other operations.

## ACID Properties

### Atomicity

**All or nothing.** If a transaction makes five writes and the system crashes after the third, atomicity guarantees that the first three are rolled back. The database is never left in a half-finished state.

Atomicity is not about concurrency (that is isolation). Atomicity is about what happens when a fault occurs during a multi-step operation.

### Consistency

**Application invariants are preserved.** If your application requires that a credit and debit always sum to zero, consistency means the database won't allow a transaction that violates this rule.

Note: Consistency is primarily an application-level property, not a database guarantee. The database can enforce certain invariants (foreign keys, unique constraints, check constraints), but most business rules must be maintained by application code.

### Isolation

**Concurrent transactions don't interfere.** Each transaction executes as if it were the only transaction running. The degree to which this is actually true depends on the isolation level.

### Durability

**Committed data is not lost.** Once a transaction commits, the data persists even if the system crashes, the power goes out, or disks fail. Implemented through write-ahead logs (WAL), replication, and backups.

**Caveat:** No durability guarantee is absolute. Multiple disk failures, correlated software bugs, or accidental deletion can still cause data loss. Durability is a spectrum, not a binary property.

---

## Isolation Levels

Isolation levels define which concurrency anomalies the database prevents. Higher isolation levels prevent more anomalies but reduce concurrency and performance.

### Read Uncommitted

**Prevents:** Nothing meaningful.
**Allows:** Dirty reads (reading data written by an uncommitted transaction).

Almost never used in practice. A transaction can read another transaction's in-progress, possibly-to-be-rolled-back writes.

### Read Committed

**Prevents:** Dirty reads, dirty writes.
**Allows:** Non-repeatable reads (reading the same row twice in one transaction yields different values because another transaction committed between the two reads).

**How it works:** Reads see only committed data. Writes only overwrite committed data. Implemented with row-level locks for writes and returning the old committed value for reads.

**Default in:** PostgreSQL, Oracle, SQL Server.

```sql
-- Transaction 1                    -- Transaction 2
BEGIN;                              BEGIN;
SELECT balance FROM accounts
  WHERE id = 1;  -- returns 100
                                    UPDATE accounts SET balance = 200
                                      WHERE id = 1;
                                    COMMIT;
SELECT balance FROM accounts
  WHERE id = 1;  -- returns 200 (non-repeatable read!)
COMMIT;
```

### Snapshot Isolation (Repeatable Read)

**Prevents:** Dirty reads, dirty writes, non-repeatable reads.
**Allows:** Write skew, phantoms.

**How it works:** Each transaction sees a consistent snapshot of the database as of the transaction's start time. Implemented with Multi-Version Concurrency Control (MVCC): the database maintains multiple versions of each row, and each transaction sees the version that was committed before the transaction started.

**Default in:** MySQL (InnoDB calls it "repeatable read"), PostgreSQL (as "repeatable read", which is actually snapshot isolation).

```sql
-- Transaction 1 sees a frozen snapshot
BEGIN;
SELECT balance FROM accounts WHERE id = 1;  -- returns 100
-- Even if another transaction changes balance to 200 and commits,
-- Transaction 1 still sees 100 for the rest of its lifetime
SELECT balance FROM accounts WHERE id = 1;  -- still returns 100
COMMIT;
```

**Key benefit:** Long-running reads (analytics, backups) don't block writes, and writes don't block reads.

### Serializable

**Prevents:** All concurrency anomalies.
**Guarantees:** Transactions execute as if they ran one after another, in some serial order.

Three implementation approaches:

#### Actual Serial Execution

Run all transactions on a single CPU core, one at a time.

**Strengths:** No concurrency bugs possible; simple implementation.
**Weaknesses:** Limited to single-core throughput; transactions must be short; no multi-statement interactive transactions.
**Used by:** VoltDB, Redis (single-threaded command execution).

#### Two-Phase Locking (2PL)

Readers block writers, and writers block readers. A transaction acquires locks as it reads and writes; it releases all locks only at commit or abort.

**Strengths:** Provides true serializability.
**Weaknesses:** Poor performance under contention; deadlocks are possible and require detection/resolution; can severely limit concurrency.
**Used by:** MySQL (InnoDB serializable mode), DB2.

```
Transaction A: Acquires shared lock on row X (for reading)
Transaction B: Wants exclusive lock on row X (for writing) -> BLOCKED
Transaction A: Commits -> releases lock
Transaction B: Acquires exclusive lock, proceeds
```

**Deadlock example:**
```
Transaction A: Lock row 1, then wants to lock row 2
Transaction B: Lock row 2, then wants to lock row 1
-> Deadlock! Database must abort one transaction.
```

#### Serializable Snapshot Isolation (SSI)

An optimistic approach: transactions execute without blocking, and the database checks for conflicts at commit time. If a conflict is detected, one transaction is aborted and retried.

**Strengths:** No blocking; reads never block writes; good performance under low contention.
**Weaknesses:** Under high contention, many transactions are aborted and retried, wasting work.
**Used by:** PostgreSQL (serializable mode since 9.1), CockroachDB.

---

## Write Skew and Phantoms

### Write Skew

Write skew occurs when two transactions read the same data, make decisions based on it, and write to different records. No single row is written by both transactions, so row-level locks don't prevent it.

**Classic example: On-call doctors**

```sql
-- Rule: At least one doctor must be on call at all times
-- Currently: Alice and Bob are both on call

-- Alice's transaction                  -- Bob's transaction
BEGIN;                                  BEGIN;
SELECT count(*) FROM doctors            SELECT count(*) FROM doctors
  WHERE on_call = true;                   WHERE on_call = true;
-- count = 2, safe to remove one        -- count = 2, safe to remove one
UPDATE doctors SET on_call = false      UPDATE doctors SET on_call = false
  WHERE name = 'Alice';                   WHERE name = 'Bob';
COMMIT;                                 COMMIT;

-- Result: NO doctors on call! Invariant violated.
```

Both transactions read count=2, both decided it was safe to go off-call, both committed. Neither wrote to the same row, so no row-level conflict was detected.

### Phantoms

A phantom occurs when a transaction's write changes the result set of another transaction's query. The write creates or removes rows that match the other transaction's WHERE clause.

**Example: Meeting room booking**

```sql
-- Transaction A                       -- Transaction B
BEGIN;                                  BEGIN;
SELECT count(*) FROM bookings           SELECT count(*) FROM bookings
  WHERE room = 101                        WHERE room = 101
  AND time = '2pm';                       AND time = '2pm';
-- count = 0, room is free              -- count = 0, room is free
INSERT INTO bookings                    INSERT INTO bookings
  (room, time, user)                      (room, time, user)
  VALUES (101, '2pm', 'Alice');           VALUES (101, '2pm', 'Bob');
COMMIT;                                 COMMIT;
-- Double booking!
```

### Preventing Write Skew

| Approach | How It Works | Limitation |
|----------|-------------|------------|
| **Serializable isolation** | Database detects and prevents all conflicts | Performance overhead |
| **SELECT FOR UPDATE** | Locks the rows that the decision is based on | Only works if there are existing rows to lock (doesn't prevent phantoms on non-existent rows) |
| **Materializing conflicts** | Pre-create rows that can be locked (e.g., create booking slots for every room-time combination) | Ugly, application-specific, error-prone |
| **Application-level locks** | Use an external lock service (Redis, ZooKeeper) | Moves complexity out of the database; risk of lock contention |
| **Unique constraints** | Database-enforced uniqueness prevents phantom inserts | Only works for simple cases |

---

## Distributed Transactions

### Two-Phase Commit (2PC)

2PC coordinates a transaction across multiple nodes (or databases):

**Phase 1 - Prepare:**
```
Coordinator -> Node A: "Prepare to commit transaction T"
Coordinator -> Node B: "Prepare to commit transaction T"
Node A -> Coordinator: "Yes, I can commit"
Node B -> Coordinator: "Yes, I can commit"
```

**Phase 2 - Commit:**
```
Coordinator -> Node A: "Commit transaction T"
Coordinator -> Node B: "Commit transaction T"
```

If any node votes "no" in phase 1, the coordinator sends "abort" to all nodes.

### 2PC Problems

- **Blocking:** If the coordinator crashes after sending "prepare" but before sending "commit/abort," participants are stuck holding locks, unable to commit or abort, potentially forever
- **Performance:** Two network round-trips plus lock holding time; 10-100x slower than single-node transactions
- **Single point of failure:** The coordinator is a critical dependency; its failure blocks all participants
- **In-doubt transactions:** Participants that voted "yes" in phase 1 cannot safely commit or abort without hearing from the coordinator

### Alternatives to Distributed Transactions

| Alternative | How It Works | Trade-off |
|-------------|-------------|-----------|
| **Saga pattern** | Break transaction into a sequence of local transactions; each step has a compensating transaction for rollback | No atomicity guarantee; compensating actions can be complex; eventual consistency |
| **Outbox pattern** | Write to the local database and an outbox table atomically; a separate process reads the outbox and publishes events | At-least-once delivery; consumers must be idempotent |
| **Event sourcing** | Store events as the source of truth; derive state from event log | Different programming model; eventual consistency for derived views |
| **Single-partition design** | Design your data model so that related data lives on the same partition | Constrains data model; may not work for all use cases |

---

## Consensus and Distributed Agreement

### The Consensus Problem

Multiple nodes must agree on a value (e.g., who is the leader, whether a transaction should commit). Consensus must satisfy:

- **Agreement:** All nodes decide the same value
- **Validity:** The decided value was proposed by some node
- **Termination:** All non-faulty nodes eventually decide
- **Integrity:** Each node decides at most once

### Consensus Algorithms

| Algorithm | Used By | Key Feature |
|-----------|---------|-------------|
| **Paxos** | Google (Chubby, Spanner) | Mathematically proven; notoriously hard to implement correctly |
| **Raft** | etcd, CockroachDB, TiKV | Designed for understandability; leader-based with log replication |
| **Zab** | ZooKeeper | Similar to Paxos; used for ZooKeeper's atomic broadcast |
| **Viewstamped Replication** | Academic | Predecessor to Raft; similar approach |

### Practical Consensus: What You Actually Use

Most applications don't implement consensus directly. Instead, they use consensus-based services:

- **ZooKeeper/etcd:** Distributed key-value store with strong consistency; used for service discovery, leader election, distributed locks, configuration management
- **Consul:** Service mesh with consensus-based service catalog
- **Google Spanner:** Globally distributed database with external consistency (linearizability + serializable isolation) using TrueTime and Paxos

### CAP Theorem in Practice

The CAP theorem states that in the presence of a network partition (P), a system must choose between consistency (C) and availability (A). In practice:

- **CP systems:** Sacrifice availability during partitions; refuse to serve requests if they can't guarantee consistency (e.g., ZooKeeper, HBase, Spanner)
- **AP systems:** Sacrifice consistency during partitions; continue serving requests with potentially stale data (e.g., Cassandra, DynamoDB, CouchDB)

**Important nuance:** CAP is about the behavior during a network partition, which is a rare event. During normal operation, you can have both consistency and availability. The question is: what happens when the network fails?

Most real-world systems are not purely CP or AP. They offer tunable consistency, where the application can choose per-operation whether to prioritize consistency or availability.