Distributed Systems

In a distributed system, you can only guarantee TWO of three properties:

C - Consistency: Every read receives the most recent write
A - Availability: Every request receives a response (success or failure)
P - Partition Tolerance: System continues to operate despite network partitions

Since network partitions are inevitable in distributed systems,
the real choice is between CP and AP:

CP Systems (Consistency + Partition Tolerance):
- During a partition, the system may become unavailable
- Examples: ZooKeeper, HBase, MongoDB (with majority read concern)
- Use when: Financial transactions, inventory management, configuration

AP Systems (Availability + Partition Tolerance):
- During a partition, the system may return stale data
- Examples: Cassandra, DynamoDB, CouchDB
- Use when: Social media feeds, product catalog, analytics

Important Nuance:
CAP is about behavior DURING a partition. When the network is healthy,
you can have all three. The question is: what do you sacrifice when
things go wrong?

If there is a Partition (P):
  Choose between Availability (A) and Consistency (C)
Else (E), when system is running normally:
  Choose between Latency (L) and Consistency (C)

Examples:
- DynamoDB: PA/EL (Available during partition, Low latency normally)
- MongoDB: PC/EC (Consistent during partition, Consistent normally)
- Cassandra: PA/EL (Available during partition, Low latency normally)
- PostgreSQL (single): N/A/EC (Not distributed, Consistent normally)

1. The network is reliable          → Build retry logic, circuit breakers
2. Latency is zero                  → Design for latency, async where possible
3. Bandwidth is infinite            → Compress, batch, paginate
4. The network is secure            → Encrypt, authenticate, authorize
5. Topology doesn't change          → Use service discovery, not hardcoded IPs
6. There is one administrator       → Design for multi-team operation
7. Transport cost is zero           → Minimize cross-region calls
8. The network is homogeneous       → Handle different protocols and formats

Purpose: Achieve consensus among distributed nodes on a shared state.

Roles:
- Leader: Handles all client requests, replicates to followers
- Follower: Replicates data from leader, votes in elections
- Candidate: Node attempting to become leader

Key Mechanisms:

1. Leader Election:
   - Followers have randomized election timeouts (150-300ms)
   - If a follower doesn't hear from leader before timeout, becomes candidate
   - Candidate requests votes from all nodes
   - Majority vote wins the election
   - New leader starts sending heartbeats

2. Log Replication:
   - Client sends request to leader
   - Leader appends to its log, sends to followers
   # ... (condensed) ...
- Each election increments the term number
- Nodes reject requests from leaders with stale term numbers
- Prevents split-brain scenarios

Example Flow:
Client → Leader (Term 5) → Append to log → Replicate to 2/4 followers
        → Majority achieved (3/5) → Commit → Respond to client

Roles:
- Proposer: Proposes values
- Acceptor: Accepts or rejects proposals
- Learner: Learns the decided value

Two Phases:

Phase 1 (Prepare):
  Proposer → Acceptors: "Prepare(n)" (n = proposal number)
  Acceptors → Proposer: "Promise(n)" if n > any previously promised

Phase 2 (Accept):
  Proposer → Acceptors: "Accept(n, value)"
  Acceptors → Proposer: "Accepted(n, value)" if still valid

Value is chosen when majority of acceptors have accepted.

Why Raft over Paxos:
- Raft is easier to understand and implement correctly
- Raft has a clear leader (simpler client interaction)
- Paxos allows multiple proposers (more complex, potential livelock)
- Most production systems use Raft (etcd, CockroachDB, TiKV)

Coordinator manages a transaction across multiple participants.

Phase 1 (Prepare/Vote):
  Coordinator → All Participants: "Can you commit?"
  Each Participant: Prepare (acquire locks, validate)
  Each Participant → Coordinator: "Yes" or "No"

Phase 2 (Commit/Abort):
  If ALL vote "Yes": Coordinator → All: "Commit"
  If ANY vote "No":  Coordinator → All: "Abort"

┌─────────────┐         ┌─────────────┐         ┌─────────────┐
│ Coordinator │         │ Service A   │         │ Service B   │
└──────┬──────┘         └──────┬──────┘         └──────┬──────┘
       │──── Prepare ────────>│                        │
       │──── Prepare ──────────────────────────────>│
       │<─── Vote Yes ───────│                        │
       │<─── Vote Yes ─────────────────────────────│
       │──── Commit ─────────>│                        │
       │──── Commit ───────────────────────────────>│
       │<─── Ack ────────────│                        │
       │<─── Ack ──────────────────────────────────│

Problems with 2PC:
- Blocking: If coordinator crashes after Phase 1, participants hold locks forever
- Single point of failure: Coordinator
- Performance: Holding locks during the entire protocol
- Not suitable for microservices (tight coupling)

A saga is a sequence of local transactions. Each step has a compensating
action that undoes its effect if a later step fails.

Example: E-Commerce Order

Step 1: Create Order      → Compensate: Cancel Order
Step 2: Reserve Inventory → Compensate: Release Inventory
Step 3: Charge Payment    → Compensate: Refund Payment
Step 4: Ship Order        → Compensate: Cancel Shipment

If Step 3 (Charge Payment) fails:
  → Execute compensating action for Step 2 (Release Inventory)
  → Execute compensating action for Step 1 (Cancel Order)

Two Coordination Approaches:

1. Choreography (Event-Driven):
   Each service listens for events and acts independently.
   Order Service → "OrderCreated" event
   # ... (condensed) ...
   A saga orchestrator tells each service what to do.
   Orchestrator → Order Service: "Create Order"
   Orchestrator → Inventory Service: "Reserve Inventory"
   Orchestrator → Payment Service: "Charge Payment"

   Pro: Easy to understand, clear flow, centralized error handling
   Con: Orchestrator is a single point of complexity

1. Design compensating actions for EVERY step
2. Compensating actions must be idempotent (safe to retry)
3. Store saga state persistently (survive crashes)
4. Handle partial compensation failures (retry with backoff)
5. Make steps as independent as possible
6. Consider semantic locks (mark resources as "pending" during saga)
7. Use correlation IDs to track the entire saga flow

Simple hashing: node = hash(key) % N
When N changes (add/remove node), MOST keys get remapped.
Example: 3 nodes → 4 nodes remaps ~75% of keys

Consistent hashing: Only K/N keys get remapped (K = total keys, N = nodes)
Adding 1 node to 3 remaps only ~25% of keys

1. Hash both nodes and keys onto a circular ring (0 to 2^32)
2. Each key is assigned to the first node clockwise from its hash position

Ring (simplified):
        Node A (hash: 100)
       /                    \
Key X (hash: 80)     Key Y (hash: 250)
      |                      |
      → Assigned to A        → Assigned to B
       \                    /
        Node B (hash: 300)

Adding Node C at hash 200:
- Keys between 100 and 200 move from B to C
- All other keys stay put

Virtual Nodes (VNodes):
- Problem: With few physical nodes, distribution is uneven
- Solution: Each physical node maps to multiple virtual positions on the ring
- Node A → vnode_A1 (hash: 50), vnode_A2 (hash: 180), vnode_A3 (hash: 320)
- More virtual nodes = more even distribution
- Typical: 100-200 virtual nodes per physical node

Gossip (epidemic) protocols spread information like a rumor:

1. Each node periodically picks a random peer
2. They exchange state information
3. Both nodes update their knowledge

Properties:
- Eventually consistent: All nodes converge to same state
- Fault tolerant: Works even when nodes fail
- Scalable: O(log N) rounds to disseminate to N nodes
- Decentralized: No single point of failure

Example (Membership):
Time 0: Node A knows about {A, B}
        Node B knows about {A, B, C}
        Node C knows about {C}

Time 1: A gossips with B → A now knows {A, B, C}
Time 2: C gossips with A → C now knows {A, B, C}
# ... (condensed) ...
2. Rumor mongering: Nodes spread only new updates
   Pro: Efficient
   Con: May not converge (rumor can "die" before reaching all nodes)

3. Aggregation: Nodes compute global aggregate (sum, average, count)
   Pro: Decentralized computation
   Con: Approximate results

Physical clocks are unreliable in distributed systems:
- Clock skew between machines
- NTP synchronization is imperfect
- Events on different machines can't be ordered by wall clock

Logical clocks provide a partial ordering of events.

Rules:
1. Before each event, increment local counter
2. When sending a message, include the counter
3. When receiving a message, set counter = max(local, received) + 1

Node A:      1 ----→ 2 ----→ 3
             |               ↑
             send(1)         receive(max(3,4)+1=5)
             ↓               |
Node B:      1 ----→ 2 ----→ 4

Limitation: Lamport timestamps can't detect concurrent events.
If a=2 and b=4, you know b happened after a.
But if a=3 and b=3, you can't tell if they're concurrent or causally related.

Each node maintains a vector of counters (one per node).

Node A: [A:0, B:0] → [A:1, B:0] → sends to B → [A:2, B:0]
Node B: [A:0, B:0] → [A:0, B:1] → receives → [A:1, B:2]

Comparison Rules:
V1 < V2 if: ALL elements of V1 ≤ V2, and at least ONE element is strictly less
V1 || V2 (concurrent) if: Some V1 elements > V2 AND some V2 elements > V1

Example:
[A:2, B:1] < [A:3, B:2]  → causally ordered (second happened after first)
[A:2, B:1] || [A:1, B:2] → concurrent (neither caused the other)

Conflict Resolution for Concurrent Events:
1. Last-writer-wins (LWW): Use physical timestamp as tiebreaker (may lose data)
2. Application-specific merge: Let the application resolve (e.g., merge shopping carts)
3. CRDTs: Data structures that automatically merge concurrent updates

- Coordinate distributed operations (one writer, many readers)
- Avoid conflicts (only one node processes a particular task)
- Ensure ordering (leader sequences events)

Using ZooKeeper:
1. All candidates create ephemeral sequential znodes under /election/
   /election/node-0000000001 (Node A)
   /election/node-0000000002 (Node B)
   /election/node-0000000003 (Node C)
2. Node with the lowest sequence number is the leader
3. Other nodes set a watch on the node with the next lower number
4. When the leader dies, its ephemeral node is deleted
5. The next node detects the deletion and becomes leader

Using etcd:
1. Candidates try to create a key with a lease (TTL)
2. First one to create it wins (atomic create)
3. Others watch the key
4. Leader must refresh the lease periodically (heartbeat)
5. If leader fails to refresh, lease expires, new election begins

Problem: Old leader doesn't know it's no longer leader (network partition,
GC pause, slow response). It continues acting as leader → split brain.

Solution: Fencing tokens
1. Each leader election produces a monotonically increasing token number
2. All requests from the leader include the fencing token
3. Resources (databases, storage) reject requests with old tokens

Leader 1 (token: 34) → Database → Accepts (34 ≥ last seen token 33)
[Leader 1 pauses, Leader 2 elected]
Leader 2 (token: 35) → Database → Accepts (35 ≥ 34)
[Leader 1 resumes, thinks it's still leader]
Leader 1 (token: 34) → Database → REJECTS (34 < 35)

Simple Lock (single Redis instance):
SET lock_key unique_value NX PX 30000
  NX: Only set if not exists
  PX: Expire after 30 seconds
  unique_value: Random UUID (for safe release)

Release Lock:
if GET lock_key == unique_value:
    DEL lock_key
(Use Lua script for atomicity)

Redlock (multiple Redis instances, safer):
1. Get current time
2. Try to acquire lock on N/2+1 Redis instances (majority)
3. If majority acquired and total time < lock TTL: lock acquired
4. If failed: release lock on ALL instances

Lock Considerations:
- Always use TTL (prevent deadlocks if holder crashes)
- Use unique token for safe release (prevent releasing someone else's lock)
- Retry with random backoff on failure
- Consider fencing tokens for stronger safety

Sometimes you can avoid distributed locks entirely:

1. Idempotent operations: Design operations that are safe to execute multiple times
2. Optimistic concurrency: Use version numbers, retry on conflict
3. CRDTs: Data structures that don't need coordination
4. Single-writer principle: Route all writes for a key to one node
5. Queue-based ordering: Instead of locking, serialize through a queue

Strongest ←──────────────────────────────────→ Weakest

Linearizability → Sequential → Causal → Eventual

Linearizability:
  - Operations appear to happen at a single point in time
  - All nodes agree on the order
  - Highest latency, lowest throughput

Sequential Consistency:
  - Operations from each node appear in order
  - Different nodes may see different global orderings
  - Common in multi-threaded programming

Causal Consistency:
  - Causally related operations appear in order
  - Concurrent operations may appear in any order
  - Good balance of consistency and performance

Eventual Consistency:
  - All replicas converge to the same state eventually
  - No ordering guarantees during convergence
  - Highest throughput, lowest latency
  - Most common in distributed systems

1. Last Writer Wins (LWW):
   - Highest timestamp wins
   - Simple but may lose data
   - Used by: Cassandra, DynamoDB (default)

2. First Writer Wins:
   - Only accept the first write, reject conflicts
   - Simple but may reject valid writes

3. Application-Level Resolution:
   - Store all conflicting versions
   - Application (or user) resolves the conflict
   - Used by: Amazon shopping cart

4. CRDTs (Conflict-free Replicated Data Types):
   - Data structures that automatically merge without conflicts
   - Types: G-Counter, PN-Counter, G-Set, OR-Set, LWW-Register
   - Used by: Redis CRDT, Riak

5. Operational Transformation:
   - Transform concurrent operations to maintain consistency
   - Used by: Google Docs, collaborative editing

[ ] Failure modes identified for every component
[ ] Retry logic with exponential backoff
[ ] Circuit breakers on all external calls
[ ] Idempotency keys for all mutating operations
[ ] Timeouts on all network calls (connect + read)
[ ] Health checks for all services
[ ] Consensus mechanism chosen (if needed)
[ ] Consistency model documented for each data store
[ ] Conflict resolution strategy defined
[ ] Network partition handling defined
[ ] Monitoring for replication lag
[ ] Clock synchronization strategy (NTP + logical clocks)