Distributed Systems Fundamentals & Core Invariants

Target Concept: Core distributed computing theories, models, failure modes, and concurrency paradigms expected in Amazon SDE II (L5) and Senior SDE (L6) system design interviews.

---

1. The Core Theorems: CAP & PACELC

A. CAP Theorem in Practice

In an asynchronous network where network partitions ($P$) are an unavoidable physical reality, a distributed storage system can guarantee at most one of:

• Consistency ($C$): Every read receives the most recent write or an error (Linearizability).
• Availability ($A$): Every non-failing node returns a non-error response for every request, without guaranteeing it contains the most recent write.

Synthesizing vector architecture diagram...

Real-World Interview Nuance (The AWS Reality)

• No Pure "CA" Systems: A network partition is not a design choice; it is a physical condition caused by fiber cuts, switch failures, or GC pauses. Systems claiming to be "CA" simply crash or become completely unavailable when a partition occurs.
• Granular CAP: Systems are rarely globally $CP$ or $AP$. For example, Amazon DynamoDB allows the caller to choose per read operation:
  • ConsistentRead: false (default) $\rightarrow$ $AP$ behavior (returns from nearest storage replica with 1 read capacity unit for 4KB, latency $<5\text{ms}$).
  • ConsistentRead: true $\rightarrow$ $CP$ behavior (queries leader/quorum replica, costs 2 read capacity units, guarantees linearizability).

---

B. PACELC Theorem (The Latency Dimension)

The CAP theorem only describes behavior during a network partition. But partitions are rare (typically $<0.1\%$ of normal operating time). What happens during normal operation?

The PACELC Theorem extends CAP:

$$\text{If } \mathbf{P} \text{ (Partition): Choose } \mathbf{A} \text{ or } \mathbf{C}; \quad \mathbf{E} \text{ (Else): Choose } \mathbf{L} \text{ (Latency) or } \mathbf{C} \text{ (Consistency)}$$

---

2. Partitioning, Sharding & Consistent Hashing

When data or throughput exceeds the physical limits of a single machine (CPU, RAM, disk I/O, network bandwidth), the dataset must be partitioned.

A. Partitioning Strategies

Synthesizing vector architecture diagram...

1. Range Partitioning:
   • Keys are sorted and partitioned into continuous ranges (e.g., timestamps, alphabetical user IDs).
   • Pro: Efficient range queries (SELECT * WHERE timestamp BETWEEN t1 AND t2).
   • Con (Hotspots): Sequential writes (e.g., current timestamp) bombard a single shard while others sit idle.
2. Hash Partitioning (Modulo Sharding):
   • $\text{Shard} = \text{hash}(\text{key}) \pmod N$.
   • Pro: Uniform data distribution across $N$ nodes.
   • Con (Resharding Catastrophe): If $N$ changes to $N+1$, nearly $100\%$ of keys hash to different nodes, requiring a massive cluster rebalance.
3. Consistent Hashing with Virtual Nodes:
   • Maps both nodes and keys to a continuous $2^{32}-1$ or $2^{128}-1$ hash ring.
   • A key belongs to the first node encountered clockwise on the ring.
   • Virtual Nodes (Vnodes): Each physical node is assigned 100–300 virtual positions on the ring.
   • Trade-off Resolution: When a node is added or removed, only $\frac{K}{N}$ keys must migrate (where $K$ is total keys, $N$ is number of nodes), and Vnodes ensure the migrated load is evenly distributed across all remaining nodes rather than overloading a single neighbor.

---

3. Data Replication & Consensus

Replication provides fault tolerance, high availability, and read scalability.

A. Replication Models

Synthesizing vector architecture diagram...

• Single-Leader Synchronous: Primary waits for all replicas to acknowledge before confirming write to client.
  • Trade-off: Zero data loss (RPO = 0), but write latency is tied to the slowest replica; write availability drops if any replica hangs.
• Single-Leader Semi-Synchronous (Amazon Aurora Model):
  • Primary replicates to 6 storage nodes across 3 Availability Zones (2 per AZ).
  • Aurora requires a 4-of-6 write quorum to commit and a 3-of-6 read quorum for recovery.
  • Trade-off: Resilient against an entire AZ failure plus an additional storage node failure without losing write availability or data.
• Multi-Leader / Active-Active:
  • Writes accepted in multiple datacenters/regions simultaneously (e.g., DynamoDB Global Tables).
  • Trade-off: Ultra-low local write latency, but requires asynchronous replication and conflict resolution (Last-Write-Wins based on physical clocks or CRDTs - Conflict-free Replicated Data Types).

---

4. Consistency Spectrum

In an interview, do not treat consistency as a binary choice between "Strong" and "Eventual". Demonstrate mastery across the spectrum:

Synthesizing vector architecture diagram...

1. Linearizability (Strict Consistency): Global real-time clock ordering. If Client A finishes writing at $T_1$, any client reading at $T > T_1$ MUST see that write.
2. Causal Consistency: Operations that are causally related must be seen by everyone in the same order. Operations that are concurrent may be observed in different orders.
3. Read-Your-Writes (Session Consistency): A user will always see updates they submitted themselves, preventing jarring UI experiences (e.g., posting a comment and refreshing, only to see it missing).
4. Monotonic Reads: Once a client reads a particular state, they will never observe an earlier state on subsequent reads (preventing "time travel" when talking to lagging read replicas).
5. Eventual Consistency: Given no new updates, all replicas eventually converge to the same value. Lowest latency and highest availability.

---

5. Distributed Transactions & Concurrency Control

A. Two-Phase Commit (2PC) vs. The Saga Pattern

Synthesizing vector architecture diagram...

---

6. Asynchronous Messaging, Streaming & Delivery Guarantees

Synthesizing vector architecture diagram...

A. Message Delivery Semantics

1. At-Least-Once (Standard SQS, Kafka default):
   • Guaranteed delivery, but network retries may cause duplicate messages.
   • Mandatory Architecture Requirement: Every consumer MUST be idempotent.
2. At-Most-Once:
   • Messages are never duplicated, but may be dropped under network failure.
   • Use case: Non-critical telemetry, metrics sampling, gaming position ticks.
3. Exactly-Once (SQS FIFO, Kafka with transactional producer):
   • Message is delivered and processed once per unique deduplication ID.
   • Trade-off: SQS FIFO throughput is capped at 3,000 msg/sec with batching, compared to virtually unlimited throughput with Standard SQS.

B. Poison Pills & Backpressure

• Dead-Letter Queue (DLQ): When a consumer fails to process a message $M$ after $K$ retries (e.g., malformed payload or unhandled exception), the message is moved to a DLQ rather than blocking the queue forever.
• Backpressure: Prevents fast producers from overwhelming slow consumers:
  • Pull-based consumers (consumer pulls when ready).
  • Rate-limiting and circuit breakers at the boundary.
  • Buffer queues with scaling policies (e.g., CloudWatch metric ApproximateNumberOfMessagesVisible triggering ECS Auto Scaling).