Distributed Databases & CAP Theorem

// WRITE PATH:
Client → Leader → [process write, append to replication log]
Leader → [send log entries to Follower1, Follower2, Follower3]
Follower1,2,3 → [apply log entries, acknowledge to leader]
Leader → [acknowledge to client when quorum confirms]

// READ PATH (different consistency levels):
// Read from leader: always linearizable (most recent data)
// Read from follower: may be stale by replication lag amount

// REPLICATION LAG:
// Asynchronous replication (PostgreSQL streaming default):
//   Leader acknowledges write immediately after writing to its own WAL
//   Followers apply changes in background — may be 0ms to seconds behind
//   Advantage: no write latency penalty
//   Disadvantage: if leader crashes before follower catches up → data loss
//
// Synchronous replication (PostgreSQL synchronous_standby_names):
//   Leader waits for at least one follower to confirm receipt before ack to client
//   Advantage: zero data loss on leader failure (failover to confirmed follower)
//   Disadvantage: write latency = leader processing + one network round-trip to follower
//   If synchronous follower is slow or unreachable → writes stall

// FAILOVER: what happens when the leader dies
// 1. Detect leader failure (heartbeat timeout, typically 10-30 seconds)
// 2. Elect new leader from the followers (choose one with most recent data)
// 3. Reconfigure clients to write to new leader
// 4. Former leader (if it recovers) must accept new leader — not try to reassert

// SPLIT-BRAIN PROBLEM:
// Old leader recovers but thinks it's still leader
// New leader was elected — both are now accepting writes
// Two nodes both believe they are the leader → data diverges catastrophically
// Solution: STONITH (Shoot The Other Node In The Head) — force old leader offline
//           Fencing tokens — writes require a monotonically increasing token
//           Quorum — leader requires votes from majority of nodes to stay leader

// POSTGRESQL SINGLE-LEADER SETUP:
-- On primary:
wal_level = replica
max_wal_senders = 10
-- On replica:
primary_conninfo = 'host=primary-ip port=5432 user=replicator'
-- Add hot_standby = on to allow read queries on replica
hot_standby = on

// WRITE CONFLICT PROBLEM:
// User A on Node 1: UPDATE posts SET title = 'Version A' WHERE post_id = 1
// User B on Node 2: UPDATE posts SET title = 'Version B' WHERE post_id = 1
// Both writes succeed locally. Both replicate to the other node.
// Node 1 sees: A then B (conflict)
// Node 2 sees: B then A (conflict)
// Which value wins? Different nodes have different orderings.

// CONFLICT RESOLUTION STRATEGIES:

// 1. Last Write Wins (LWW):
// Attach a timestamp to every write. The write with the latest timestamp wins.
// Problem: clock skew between nodes means "latest" is unreliable.
//   Node1 clock is 1ms ahead of Node2 → Node1's writes always win
//   DynamoDB, Cassandra, and others use LWW by default — can silently drop data.

// 2. Version Vectors / Vector Clocks:
// Each value carries a version vector: {Node1: version, Node2: version, ...}
// On write: increment this node's version number
// On conflict: if neither version vector "dominates" the other → concurrent writes
// Application must merge or choose between conflicting versions
// Used by Riak, Amazon's Dynamo paper

// 3. Merge Functions (CRDTs):
// Design data types that automatically resolve conflicts mathematically
// Counter: value = sum of all increments from all nodes (always correct)
// Set: union of all additions, tombstones for deletions
// G-Counter, PN-Counter, LWW-Register, OR-Set are common CRDTs
// Cassandra uses CRDTs for certain data types

// 4. Application-Level Resolution:
// Store all conflicting versions
// Present conflict to application (e.g., git merge conflict)
// Application logic decides what to keep
// CouchDB's document revision system works this way

// WHEN TO USE MULTI-LEADER:
// Multiple data centres that must accept writes locally
// Offline-capable applications (sync when reconnected)
// Collaborative editing (Google Docs — every keystroke is a concurrent write)

// TOOLS: Galera Cluster (MySQL), BDR (PostgreSQL), CockroachDB

// QUORUM SYSTEM:
// N = total number of replica nodes
// W = number of nodes that must confirm a write (write quorum)
// R = number of nodes that must respond to a read (read quorum)
// Rule: W + R > N guarantees at least one node in every read has the latest write

// EXAMPLE: N=3, W=2, R=2 (Cassandra QUORUM level)
// Write to all 3 nodes, wait for 2 to confirm: W=2
// Read from all 3 nodes, wait for 2 to respond: R=2
// W + R = 4 > N=3 → there is guaranteed overlap of at least 1 node
// That overlapping node has the latest write → read returns the latest value

// AVAILABILITY vs CONSISTENCY TRADE-OFF:
// W=1, R=1 (N=3): ultra-low latency, no consistency guarantee
//   One write succeeds → client gets ack. One read responds → client gets data.
//   W+R=2 < N=3 → no guaranteed overlap → can read stale value

// W=3, R=1 (N=3): every write requires all 3 nodes
//   Highly durable (all nodes have the write) but one slow/down node blocks writes
//   R=1: any one node can answer reads — but it always has the latest (W=N)

// W=2, R=2 (N=3): balanced — quorum reads and writes
//   W+R=4>N=3 → guaranteed overlap → linearizable
//   One node can be down and both reads and writes succeed (W=2, R=2, healthy=2≥2)

// W=1, R=3 (N=3): fast writes, reads check all nodes
//   W+R=4>N=3 → linearizable  but read must check ALL nodes

// CASSANDRA CONSISTENCY LEVELS:
// ONE:     W=1, R=1  → fastest, weakest consistency
// QUORUM:  W=ceil(N/2+1), R=ceil(N/2+1) → balanced
// ALL:     W=N, R=N  → slowest, strongest consistency
-- Set per-query in Cassandra:
-- SELECT * FROM orders WHERE order_id=1 USING CONSISTENCY QUORUM;
-- INSERT INTO orders (...) VALUES (...) USING CONSISTENCY ONE;

// READ REPAIR: when a read returns stale data from some nodes
// The coordinator detects inconsistency (different versions from R nodes)
// Sends updated value to the nodes that had stale data
// Eventual consistency progresses toward consistency through read traffic

// HINTED HANDOFF: when a target node is down during a write
// The write is accepted by another node as a "hint"
// When the target recovers, the hinting node delivers the stored write
// Ensures writes are not lost even when target is temporarily unavailable

// PROBLEM: Join customers (sharded by city) with orders (sharded by date)
// Query: SELECT c.name, o.total FROM customers c JOIN orders o ON c.id=o.customer_id
// WHERE c.city='Bengaluru' AND o.order_date >= '2024-01-01'

// Strategy 1: Scatter-Gather
// Send query to all shards, collect results, join in the application layer
// Application layer becomes the join processor
// O(N_shards²) network calls → terrible at scale

// Strategy 2: Co-location (best strategy when possible)
// Shard BOTH tables by the same key used for the join (customer_id)
// customer_id=1001 → Shard 3 for customers AND Shard 3 for orders
// Join between customer 1001 and their orders is entirely within Shard 3
// The join is local — no cross-shard communication

// Implementation: consistent hash on customer_id for BOTH tables
// customer table: hash(customer_id) → shard
// orders table:   hash(customer_id) → SAME shard  ← key: shard by the join key
// All queries "by customer" are entirely local to one shard

// Strategy 3: Denormalisation
// Store the join data redundantly: include customer_name in the orders table
// orders.customer_name = redundant copy of customers.name
// "Join" is just reading orders — no cross-shard lookup needed
// Cost: update anomalies (customer name change → update all their orders)

// Strategy 4: Broadcast joins
// One table is small enough to broadcast to all shards
// Restaurants table: 10,000 rows — tiny
// Orders table: 5 billion rows — large and sharded
// Strategy: keep a full copy of restaurants on EVERY shard
// Join restaurants × orders on each shard → combine results
// Works when one side of the join is small (dimension table in star schema)

// PARTICIPANTS: Coordinator C, Nodes N1, N2, N3
// Transaction T wants to commit — it modified data on N1, N2, and N3

// ─────────────────────────────────────────────────────────────────
// PHASE 1: PREPARE (Voting Phase)
// ─────────────────────────────────────────────────────────────────

// Coordinator sends PREPARE(T) to all participants
C → N1: "PREPARE transaction T"
C → N2: "PREPARE transaction T"
C → N3: "PREPARE transaction T"

// Each participant:
// 1. Checks if it CAN commit (has all necessary locks, no constraint violations)
// 2. If YES: writes all transaction changes to its WAL (durably)
//            Acquires all necessary locks for the commit
//            Sends VOTE-COMMIT to coordinator
// 3. If NO: sends VOTE-ABORT to coordinator
// KEY PROPERTY: once a node sends VOTE-COMMIT, it has promised to commit
//               if the coordinator later says COMMIT. It cannot change its mind.
//               It must hold its locks and preserve its prepared state until
//               the coordinator tells it what to do.

N1 → C: "VOTE-COMMIT"
N2 → C: "VOTE-COMMIT"
N3 → C: "VOTE-COMMIT"

// ─────────────────────────────────────────────────────────────────
// PHASE 2: COMMIT (Decision Phase)
// ─────────────────────────────────────────────────────────────────

// Coordinator decision:
// IF all votes are COMMIT → send COMMIT to all participants
// IF ANY vote is ABORT → send ABORT to all participants

// All voted COMMIT:
C writes COMMIT(T) to its own WAL (durable — now committed to committing)
C → N1: "COMMIT transaction T"
C → N2: "COMMIT transaction T"
C → N3: "COMMIT transaction T"

// Each participant:
// Applies the commit (writes committed state, releases locks)
// Sends ACK to coordinator
// Cleans up prepared state

N1,N2,N3 → C: "ACK"
// Transaction complete ✓

// ─────────────────────────────────────────────────────────────────
// FAILURE SCENARIOS
// ─────────────────────────────────────────────────────────────────

// SCENARIO 1: Node fails BEFORE sending VOTE-COMMIT
// N2 crashes before responding to PREPARE
// Coordinator times out waiting for N2's vote
// Coordinator sends ABORT to N1 and N3
// When N2 recovers: no prepared state → aborts
// Result: Transaction aborted ✓ (no partial commit)

// SCENARIO 2: Node fails AFTER sending VOTE-COMMIT (the blocking problem)
// N2 sent VOTE-COMMIT then crashed
// N2 holds locks and has prepared state on disk
// N2 recovers: "I said COMMIT but don't know if coordinator decided COMMIT or ABORT"
// N2 MUST WAIT for coordinator to tell it what to do
// N2 CANNOT release its locks — what if coordinator said COMMIT?
// N2 CANNOT commit unilaterally — what if coordinator said ABORT?
// ← THIS IS 2PC's BLOCKING PROBLEM: a prepared node blocks indefinitely
//   waiting for the coordinator if the coordinator crashes

// SCENARIO 3: Coordinator fails after sending COMMIT to N1 but not N2, N3
// N1 committed. N2, N3 are waiting for COMMIT/ABORT.
// When coordinator recovers: reads its WAL, sees COMMIT(T) was written
// Resends COMMIT to N2 and N3 → they commit
// Transaction complete ✓ (coordinator log makes recovery possible)

// THE 2PC BLOCKING PROBLEM:
// If the coordinator is unavailable AND at least one participant is in PREPARED state:
// Those participants CANNOT make progress — they must hold their locks indefinitely
// This can block the entire database in the worst case
// Solution: 3PC (Three-Phase Commit) adds a third phase to reduce blocking
//           But 3PC is complex and not widely used in practice
//           Modern systems use: Paxos/Raft-based consensus instead

// PROBLEM: 2PC across microservices is too coupled and too slow
// Each service locks its data waiting for coordinator → poor availability

// SAGA: Break a distributed transaction into a sequence of local transactions
// Each local transaction updates one service and publishes an event
// If any step fails: run compensating transactions to undo previous steps

// EXAMPLE: Swiggy order placement saga
// Step 1: OrderService.createOrder() → publishes ORDER_CREATED
// Step 2: InventoryService.reserveItems() → publishes ITEMS_RESERVED
//          OR → publishes RESERVATION_FAILED
// Step 3: PaymentService.processPayment() → publishes PAYMENT_COMPLETED
//          OR → publishes PAYMENT_FAILED
// Step 4: DeliveryService.assignDelivery() → publishes DELIVERY_ASSIGNED

// FAILURE AND COMPENSATION:
// If PaymentService fails (PAYMENT_FAILED):
//   Compensating transaction 2: InventoryService.releaseReservation()
//   Compensating transaction 1: OrderService.cancelOrder()

// IMPLEMENTATION STYLES:
// Choreography: each service listens for events and reacts
//   No central coordinator
//   Services publish and consume events directly
//   Simple but hard to track overall progress

// Orchestration: a saga orchestrator sends commands and waits for responses
//   Central saga orchestrator knows the full workflow
//   Easier to understand and track but orchestrator is a potential bottleneck

// SAGA vs 2PC:
// 2PC: distributed ACID, holds locks, blocks on failures, tightly coupled
// Saga: eventually consistent, no long-held locks, compensating instead of rollback
//       application must design for partial completion and compensation logic
// Saga is preferred in microservices architectures
// 2PC is used within a single database cluster (e.g., CockroachDB internally)

// RAFT CLUSTER: N nodes, one leader at a time

// TERMS: Raft divides time into terms (monotonically increasing integers)
// Each term begins with a leader election
// At most one leader per term (possibly no leader if election fails)

// ─────────────────────────────────────────────────────────────────
// LEADER ELECTION
// ─────────────────────────────────────────────────────────────────
// Each node is in one of: FOLLOWER, CANDIDATE, LEADER

// FOLLOWER: waits for heartbeats from leader
// If election timeout expires without hearing from leader:
//   → become CANDIDATE, start an election

// CANDIDATE:
// 1. Increment current term
// 2. Vote for itself
// 3. Send RequestVote(term, candidateId, lastLogIndex, lastLogTerm) to all nodes
//
// A node grants a vote if:
//   a. Haven't voted in this term yet
//   b. Candidate's log is at least as up-to-date as the voter's log
//      (ensures elected leader has all committed entries)

// MAJORITY VOTE: candidate becomes LEADER when it receives votes from majority
// With N=5 nodes: need 3 votes (majority = floor(N/2) + 1 = 3)
// Majority ensures at most one leader per term (two candidates cannot
// both win majority — pigeonhole principle)

// ─────────────────────────────────────────────────────────────────
// LOG REPLICATION
// ─────────────────────────────────────────────────────────────────
// Client sends write to leader
// Leader appends entry to its log (with current term number)
// Leader sends AppendEntries(entries, prevLogIndex, prevLogTerm) to all followers
// Followers append entries to their logs and send success
// Once leader receives success from MAJORITY: entry is COMMITTED
// Leader applies committed entry to state machine, responds to client
// Leader notifies followers of commit in next AppendEntries
// Followers apply committed entries to their state machines

// KEY SAFETY PROPERTY:
// Once an entry is committed (applied by majority), it will NEVER be removed
// Future leaders will always have all committed entries (guaranteed by vote restriction)
// = RAFT SAFETY GUARANTEE

// LEADER CRASH RECOVERY:
// New leader elected by majority vote
// New leader has all committed entries (guaranteed by voting rules)
// New leader replaces followers' uncommitted entries with its own
// Uncommitted entries from old leader are lost (they were never committed = never ack'd to client)

// COMPARING RAFT vs PAXOS:
// Raft: one leader replicates log sequentially → easy to understand, implement
// Paxos: no designated leader for basic algorithm, complex multi-paxos for logs
// Paxos: mathematically elegant, complex to implement correctly
// Multi-Paxos: practical variant with a leader, closer to Raft
// Modern preference: Raft for new implementations (etcd, CockroachDB, TiKV, Consul)