System Design · HLD Interview Questions

Answer: In traditional hashing, keys are mapped to servers using node_index = hash(key) % N (where $N$ is the cluster size).

• The Problem: If $N$ changes (due to a node crash or addition), almost all keys are mapped to completely different indices. This triggers a massive, immediate cache miss storm, crashing databases.
• Consistent Hashing: Maps both keys and physical servers onto a circular 32-bit integer ring space ($0$ to $2^{32}-1$).

graph TD
    subgraph Hash Ring
        Ring[Circle 0 to 2^32-1]
        NodeA[Server A] -->|Clockwise Next| NodeB[Server B]
        NodeB -->|Clockwise Next| NodeC[Server C]
        NodeC -->|Clockwise Next| NodeA
    end
    Key[Key: hash=1200] -->|Move Clockwise| NodeB

• Routing Mechanism: A key's hash is placed on the ring. Move clockwise until the first server hash is reached; store the key on that server.
• Node Addition/Removal: Only keys located between the modified node and its counter-clockwise neighbor need re-mapping. All other nodes remain untouched.
• Virtual Nodes (VNodes): Maps a physical node to dozens of virtual positions across the ring (e.g., Server_A_1, Server_A_2). This ensures keys are distributed uniformly, preventing hot-spots on single machines.

---

Answer: Sharding splits a large database horizontally, storing different subsets of rows (shards) across distinct database servers.

graph LR
    User[User ID] --> Router{Shard Router}
    Router -->|ID: 1-100| Shard1[(Shard 1)]
    Router -->|ID: 101-200| Shard2[(Shard 2)]

• Range-Based Sharding:
  • Data is split based on ranges of a key attribute (e.g., IDs $0-10,000$ go to Shard 1, $10,001-20,000$ go to Shard 2).
  • Risk: Severe hotspots (e.g., if new users are highly active and get higher IDs).
• Key-Based (Hash-Based) Sharding:
  • Uses a hash function on the sharding key to select the shard: shard_id = hash(key) % total_shards.
  • Risk: Hard to scale total shards without re-sharding all existing data (mitigated by Consistent Hashing).
• Directory-Based Sharding:
  • Maintains a lookup table/directory indicating which shard contains what key.
  • Risk: The lookup table itself becomes a single point of failure (SPOF) and query bottleneck.
• Main Challenges:
  • Cross-Shard Joins: Performing JOIN queries across multiple servers is incredibly expensive and slow.
  • Referential Integrity: Enforcing foreign keys across shards is practically impossible.
  • Re-sharding: Splitting an existing, overloaded shard during active traffic is extremely complex.

---

Answer: The CAP Theorem states that a distributed system can guarantee at most two of the following three properties simultaneously in the event of a network partition:

graph TD
    C[Consistency - Strong CP] --- A[Availability - Highly AP]
    A --- P[Partition Tolerance - Vital]
    P --- C

• Consistency (C): Every read receives the most recent write or an error.
• Availability (A): Every non-failing node returns a non-error response (without guaranteeing it contains the most recent write).
• Partition Tolerance (P): The system continues to operate despite arbitrary packet loss or network splits between nodes.
• The Trade-off: Network partitions ($P$) are physical facts. Therefore, during a partition, a system must choose:
  • CP (Consistency over Availability): Reject writes or block reads until network splits heal, guaranteeing data correctness but degrading availability.
  • AP (Availability over Consistency): Accept reads and writes on both split sides, keeping the system fully available but risking split-brain data conflicts.

---

Answer: A Message Queue (MQ) decouples message producers and consumers, buffering request bursts.

• RabbitMQ (AMQP Push-Model):
  • Broker Internals: Messages are pushed to exchanges, routed to queues based on routing keys, and pushed to active consumers.
  • Message State: Tracked by the broker. Once a message is acknowledged, it is deleted from disk.
  • Use Case: Complex routing, transaction flows, and short-lived message processing.
• Apache Kafka (Log-Based Pull-Model):
  • Broker Internals: An append-only commit log. Messages are written sequentially to partitions.
  • Message State: Tracked by consumers via an offset marker. Messages persist on disk after consumption (immutable history).
  • Use Case: High-throughput event streaming, log aggregation, real-time analytics.

---

Answer: Throttling engines used to enforce system-wide fair-use limits.

• Token Bucket:
  • Tokens are added to a bucket at a fixed rate. Requests consume tokens. If empty, requests are dropped.
  • Pros: Natural handling of short traffic bursts.
• Leaky Bucket:
  • Requests enter a queue. The queue drips (dispatches) requests at a constant, smooth rate.
  • Pros: Smooths out traffic, preventing downstream spikes.
• Fixed Window Counter:
  • Counts requests within fixed time windows (e.g., 100/min from 12:00 to 12:01).
  • Cons: Allows double the limit at window boundaries (e.g., 100 requests at 11:59:59 and 100 at 12:00:01).
• Sliding Window Log:
  • Saves exact timestamps of all requests in a sorted set (like Redis ZSET). Evicts timestamps older than (Current Time - Window Size).
  • Cons: Extremely high memory consumption because every single request timestamp must be stored.
• Sliding Window Counter:
  • Hybrid approach. Computes a weighted sum of request counts from the current and previous windows to estimate instantaneous rate.
  • Pros: Memory efficient; smooths boundary spikes.

---

Answer: A ride-sharing system must track latitude/longitude updates for millions of active drivers and allow riders to search nearby regions instantly. Executing standard relational queries using WHERE lat BETWEEN x AND y results in full table scans ($O(N)$), crashing database processing under heavy traffic.

To solve this, we use Geospatial Indexing to map a 2D physical point into a 1D indexable sequence.

graph TD
    Geo[Geospatial Indexing] --> Geohash[Geohash: Base32 String]
    Geo --> Quadtree[Quadtree: Recursive 4-Quadrant Division]
    Geo --> S2[Google S2: Hilbert Curve Grid Cells]

• Geohashes:
  • Mechanism: Divides the Earth into grid zones, recursively splitting them into smaller regions. Each cell is encoded as a Base32 string (e.g., 9q8yy). Longer prefixes represent tighter boxes.
  • Edge/Boundary Problem: Two points right next to each other on a grid border can have completely different Geohash prefixes (e.g., one starts with 9 and the other with d), making prefix queries fail to capture immediate neighbors.
• Quadtrees:
  • Mechanism: A tree structure where each node has exactly four children (NW, NE, SW, SE). Nodes split recursively only when a quadrant exceeds capacity bounds (e.g., $\ge 100$ drivers).
  • Pros: Dynamically sizes cells (deep trees in New York, shallow in deserts).
  • Cons: Very complex to update, balance, and synchronize in real-time memory across a distributed cluster when drivers are moving fast.
• Google S2:
  • Mechanism: Projects the Earth's sphere onto a cube, mapping it to a 1D index using a mathematical Hilbert Curve. Divides the Earth into 31 hierarchical levels of cells.
  • Pros: Solves the boundary problem cleanly. Preserves spatial locality: points physically close in 2D space are highly likely to be adjacent on the 1D Hilbert index. The industry standard.

---

Answer: The CAP Theorem states that in the event of a network Partition ($P$), a system must choose between Consistency ($C$) and Availability ($A$).

Expansion to PACELC:

Partitions are rare in physical datacenters. PACELC models the system trade-offs during normal operations (when there is no partition).

$$\text{If Partition (P)} \longrightarrow \text{choose Availability (A) or Consistency (C)}$$ $$\text{Else (E)} \longrightarrow \text{choose Latency (L) or Consistency (C)}$$

graph TD
    PACELC{PACELC Theorem} --> P[If Partition: P]
    PACELC --> E[Else Normal: E]
    P --> P_A[Availability: PA]
    P --> P_C[Consistency: PC]
    E --> E_L[Latency: EL]
    E --> E_C[Consistency: EC]

• PA/EL System (e.g., MongoDB, DynamoDB): During partitions, prioritizes availability. During normal operations, prioritizes low latency by writing/reading locally and updating replicas asynchronously.
• PC/EC System (e.g., Google Spanner, standard RDBMS): Prioritizes consistency at all times. During normal operations, it blocks writes until synchronous replication across all nodes completes, increasing latency but preventing stale reads.

---

Answer: Consensus protocols guarantee that a cluster of nodes can agree on a sequence of state values, even if some nodes fail.

• Raft (Leader-Based Protocol):
  • States: Follower, Candidate, Leader.
  • Leader Election: If a follower stops receiving heartbeats (Election Timeout), it transitions to Candidate, increments the Term, votes for itself, and requests votes from peers. It becomes Leader if it receives a majority of votes ($\ge N/2 + 1$).
  • Log Replication: All writes go to the Leader. The Leader writes sequentially to its log and sends an AppendEntries RPC to followers. Once a majority of followers ACK the write, the Leader commits it and applies it to its local state machine.
  • Split-Brain Prevention: The majority requirement ensures only one candidate can secure enough votes to become Leader in any partition. If a partition heals, the leader with the lower Term is demoted to follower.
• Paxos (Leaderless/Multi-Leader):
  • Operates in two phases:
    1. Phase 1 (Prepare): Proposer sends a proposal number $N$ to acceptor nodes. Acceptors return their highest accepted value if they promise not to accept any future proposals numbered $< N$.
    2. Phase 2 (Accept): If a majority of acceptors agree, the proposer broadcasts the value to acceptors to finalize the write.

---

Answer: Standard physical clocks cannot be used to order events in distributed systems due to network latencies and inevitable clock drift.

• Lamport Timestamps:
  • Every node maintains a local integer counter, initialized to 0.
  • Before executing a local event, increment the counter: $L = L + 1$.
  • When sending a message, attach the local counter value.
  • When receiving a message, update local counter: $L_{\text{recv}} = \max(L_{\text{local}}, L_{\text{msg}}) + 1$.
  • Limitation: If $L(A) < L(B)$, we cannot conclude that event A caused event B; it only provides a total ordering, not causal history.
• Vector Clocks:
  • Every node maintains an array/vector of size $N$ (where $N$ is the cluster size).
  • When sending a message, Node $i$ increments its own element: $V_i[i] = V_i[i] + 1$ and attaches the vector.
  • When receiving a message from Node $j$, update local vector: $V_i[k] = \max(V_i[k], V_{\text{msg}}[k])$ for all $k$, and increment $V_i[i] = V_i[i] + 1$.
  • Causality: Event $A$ causally preceded $B$ if every element in $V(A) \le V(B)$ and at least one element is strictly smaller. If elements diverge (e.g., $V(A)$ has some higher values and $V(B)$ has others), a conflict is detected, requiring application-level resolution.

---

Answer: CRDTs are distributed data structures that can be updated independently and concurrently across replicas without coordination, guaranteeing eventual convergence without conflict resolution loops.

• State-Based (CvRDT):
  • Replicas sync by sending their entire local state to other replicas.
  • A merge function must be commutative ($A \sqcup B = B \sqcup A$), associative ($A \sqcup (B \sqcup C) = (A \sqcup B) \sqcup C$), and idempotent ($A \sqcup A = A$).
  • Use Case: PNCounter (Positive-Negative Counter), LWW-Element-Set (Last-Write-Wins Set).
• Operation-Based (CmRDT):
  • Replicas sync by broadcasting individual operation payloads (e.g., "Add 5", "Remove Item X").
  • Requires a reliable, causal, ordered network delivery channel to guarantee all replicas apply operations in compatible sequences.
  • Use Case: Collaborative real-time document editors (e.g., Figma, Google Docs).

---