System Design Building Blocks

Non-functional Requirements

• CAP: When a network partition occurs, do you choose availability or consistency?
  • Consistency prioritizing systems: Ticket booking (prevent double booking), e-commerce inventory (avoid overselling), financial systems, ride matching, online auctions.
  • Availability prioritizing systems: content platforms (netflix, yelp, instagram), whatsapp (messages eventually consistent), google docs (eventually consistent), distributed cache.
  • Different features within the same system may have different consistency needs.
• Scalability
  • Evaluate the read vs. write ratio. Is it a read-heavy system (like Instagram feeds, YouTube streaming) or a write-heavy system (like Uber's location tracking, ad click aggregation, write-optimized databases like Cassandra)?
  • Consider unique scaling requirements, such as bursty traffic (e.g., holiday sales, breaking news events).
• Latency: How quickly does the system need to respond to user requests, especially for computationally intensive operations?
  • Instagram: Feed load times < 500ms, instant photo/video rendering. Distributed Job Scheduler: Execute jobs within 2s of scheduled time. Robinhood: Low latency for price updates and order placement (under 200ms).
• Fault Tolerance: How well does the system need to handle failures (e.g., redundancy, failover, recovery mechanisms)?
  • Payment System: Guarantee durability and auditability, no transaction data lost. Web Crawler: Resume crawling without losing progress.
• Environment Constraints: Are there limitations like mobile devices with limited battery/memory/bandwidth (e.g., streaming video on 3G)?
• Security: How secure does the system need to be? Consider data protection, access control, and compliance.
• Compliance: Are there legal or regulatory requirements (e.g., industry standards, data protection laws)?

Scaling Reads

1. Ensure the db is organized correctly.

• Normalization and denormalization. Indexes.
  • Gotta fix the index so you avoid table scans as much as possible.
    • B Tree is the most common index. There are others: hash index, inverted index etc.
• You can get a lot out your databases by just vertically scaling: ssds over spinning disks, more RAM, faster CPUs (to handle concurrent queries).
• Normalization is the process of structuring data to reduce redundancy by splitting information across multiple tables to avoid storing duplicate data. While this saves storage space, it makes queries more complex because you need joins to bring related data back together.
• For read-heavy systems, denormalization (the opposite of normalization - you store redundant data) trades storage for speed. Instead of joining three tables to get user profile data, store the data redundantly in a single table.
• You can materialize views by precomputing expensive aggregations.

2. Scale horizontally.

1. Read replicas will help with distribute load.
2. Sharding helps in two ways: smaller datasets mean faster individual queries, and you can distribute read load across multiple databases.

3. Add external caching layers.

1. Application level caches like redis or memcached (in memory, both). "When your application needs data, it checks the cache first. On a hit, you get sub-millisecond response times. On a miss, you query the database and populate the cache for future requests."
2. CDN and edge caching. Deliver both static and dynamic content faster, better, by distributing it geographically. CDNs only make sense for data accessed by multiple users. Don't cache user-specific data like personal preferences, private messages, or account settings.

Scaling Writes

Vertical Scaling and Write Optimization

• Writes are bottlenecked by disk I/O, CPU, or network bandwidth. Fix these first.
• Pick the right database optimized for writes.
  • Example, Cassandra, which achieves this via append-only commit log architecture. No data updates in place (and hence no disk seeks), instead writes everything sequentially to disk. Gives 10,000+ writes per second, compared to maybe 1,000 writes per second for a traditional relational database doing the same work.
  • that makes it bad for reading though, so that's the trade off.

Sharding, Partitioning

• Horizontal sharding: split the writes to N servers, each which can handle M writes/sec instead of just one server with M writes/sec. Use consistent hashing to determine which key lands in which server.
• How does one determine a parition key? A good key is one which distributes the data evenly across the cluster.

Handling Bursts

Real world traffic is not steady. You will have to deal with peak volume. Two ways to deal with it: queues, load shedding.

• Use queues to handle bursty traffic. Queues allow the application layer to act as if the data is already recorded. Database processes writes at a steady rate while the queue handles bursts.
  • This only works if the application layers writes to the queue slower than the records can be written to the database.
• Load shedding is rejecting traffic altogether. Decide which writes are important for the business and which are not.

Batching and Hierarchical Aggregation

• Individual write operations have overhead like network round trips, transaction setup, index updates. Amortize the overhead by batching writes.
  • Can be done at application year, somewhere intermediate, or in database layer.
• For high volume data, like stream processing, individual events don't matter. You can aggregate data in stages because that's what is insighful for this sort of workflows. Example, comments or likes in a live streamed video.

CDC

• Change Data Capture (CDC) is a mechanism that monitors a database's write-ahead log (WAL) or oplog, capturing every committed change as an event.
• Once a change is captured, CDC publishes it as an event to an immutable event stream. The event stream acts as a central hub from which different specialized services or consumers can independently read and process the changes.
• Kafka (or similar event streams like Kinesis) is a frequently used and highly robust intermediate step. When CDC captures changes, it publishes these changes to an immutable event stream like Kafka, creating an append-only log of every state transition.
• Note, while it's frequently used, it's not necessary. The core concept is that CDC produces an event stream.
  • For instance, DynamoDB Streams is a built-in CDC mechanism that captures changes to DynamoDB tables. While one can use Kafka, they also directly support triggering AWS Lambda functions in response to changes. In this scenario, the Lambda function acts as a direct consumer of the stream, and while Lambda functions often interact with queues, the DynamoDB Stream itself serves as the event pipeline without another explicit message queue layer between the CDC stream and the immediate consumer (the Lambda).

Redis

• Redis can run in several infrastructure configurations: as a single node, with a high availability (HA) replica, or as a cluster. When operating as a cluster, Redis is a distributed system that scales horizontally by spreading data across multiple machines.
• In a Redis cluster, data is partitioned using hash slots. Instead of you manually keeping track of which partition each piece of data belongs to, clients directly connect to the specific node that contains the data they are requesting.
• Choosing how to structure your keys is how you scale Redis.
• Redis is very fast because it is an in-memory, single-threaded data structure store written in C.
  • sub-millisecond latency for reads, O(100k) operations per second.
• It supports a wide range of data structures like strings, hashes, lists, sets, sorted sets, Bloom filters, and geospatial indexes, which are easy to reason about in a distributed system.
• Distributed Locks: Useful for short-term resource locking in high-concurrency scenarios, such as ticket booking or ride-sharing matching, using atomic operations with TTLs.
• Redis prioritizes speed over strong durability guarantees, meaning some data loss is possible during failures unless explicitly configured with persistence options like AOF (Append-Only File) or RDB snapshots, or using alternatives like AWS' MemoryDB.
  • Uneven load distribution can lead to "hot key" issues, where a single key overwhelms a node.
  • While instances can handle terabytes, it's fundamentally memory-bound. If your dataset greatly exceeds available RAM and fine-grained sharding isn't feasible, other databases might be more suitable

Redis Cluster

How Sharding Works in Redis Cluster

In Redis Cluster, the data space is divided into 16,384 hash slots. Each of these slots is owned by a single Redis node. When you use Redis Pub/Sub, the channel name is hashed to determine which of the 16,384 slots it belongs to.

• Shard: A shard is a group of nodes that manages a subset of the hash slots. For example, in a 3-node cluster, Node A might be responsible for slots 0-5460, Node B for 5461-10922, and Node C for 10923-16383.
• Channel: Each Pub/Sub channel (e.g., post:123, post:456) is assigned to one of these hash slots. Since a single node (or shard) owns thousands of slots, it will manage thousands of channels, each representing a different post or data stream.

Practical Example

Let's say you have a Redis Cluster with three nodes (Node A, Node B, and Node C).

• When a new comment is posted to post:123, the Redis client hashes the channel name post:123 and determines it belongs to slot 1000. Node A owns this slot.
• When a new comment is posted to post:456, the Redis client hashes the channel name post:456 and determines it belongs to slot 6000. Node B owns this slot.

Any SSE service instance that wants to listen for updates on post:123 must connect to Node A to subscribe. Similarly, any service instance listening for post:456 must connect to Node B.

Therefore, a single Redis node within a shard is responsible for all the channels (and thus, all the posts) that hash to its assigned range of slots.

While a cluster can have many nodes, a typical production setup for high availability and fault tolerance starts with a minimum of six nodes—three master nodes and three replica (slave) nodes, with one replica for each master. This configuration ensures that if a master node fails, a replica can be promoted to take its place.

Real Time Processing

1. Server Side Processing and Dataflow

• Event Streams (Kafka/Kinesis): Use Apache Kafka or AWS Kinesis as a scalable, durable event streaming platform to absorb bursts of data and ensure messages are processed in order using partitions.
• Stream Processors (Flink/Spark Streaming): Tools like Apache Flink or Spark Streaming process events from the stream in real-time. They can maintain in-memory aggregations and flush results to a database, enabling low-latency analytics.
• Precomputation and Denormalization: For very low-latency queries (e.g., 10s of milliseconds), data is often precomputed and stored.

2. Client Side Update Mechanisms

• Server-Sent Events (SSE): Best for unidirectional pushes from server to client. It uses a single, persistent HTTP connection for streaming data. SSE is simple to implement, has automatic reconnection, and is efficient for one-way updates like live stock prices or comments.
• WebSockets: Ideal for bi-directional, high-frequency communication by upgrading an HTTP connection to a full-duplex channel.
• WebRTC: Enables direct peer-to-peer communication for applications like video/audio calls, but is the most complex.

3. Server-to-Client Update Propagation

• Pub/Sub: A flexible pattern for broadcasting updates. A server publishes an event to a topic, and all subscribed client-facing servers receive it, then forward to their connected clients.
• Consistent Hashing: Used for stateful connections where a client's requests are always routed to the same server (e.g., in Google Docs, all editors for a document connect to the same server). It ensures optimal resource use and allows dynamic scaling.

Kafka

• A Kafka cluster is made up of multiple brokers. These are just individual servers (they can be physical or virtual). Each broker is responsible for storing data and serving clients. The more brokers you have, the more data you can store and the more clients you can serve.
• Each broker has a number of partitions. Each partition is an ordered, immutable sequence of messages that is continually appended to.
• A topic is just a logical grouping of partitions. Topics are the way you publish and subscribe to data in Kafka. When you publish a message, you publish it to a topic, and when you consume a message, you consume it from a topic. Topics are always multi-producer; that is, a topic can have zero, one, or many producers that write data to it.
• Topics are just a way to organize your data, while partitions are a way to scale your data.
• Producers are the ones who write data to topics, and consumers are the ones who read data from topics. While Kafka exposes a simple API for both producers and consumers, the creation and processing of messages is on you, the developer. Kafka doesn't care what the data is, it just stores and serves it.
• When a message is published to a Kafka topic, Kafka first determines the appropriate partition for the message. This partition selection is critical because it influences the distribution of data across the cluster.
  • Once the partition is determined, Kafka then identifies which broker holds that particular partition. The mapping of partitions to specific brokers is managed by the Kafka cluster metadata, which is maintained by the Kafka controller (a role within the broker cluster).
• Each partition in Kafka functions essentially as an append-only log file. Messages are sequentially added to the end of this log, which is why Kafka is commonly described as a distributed commit log.
  • Once written, messages in a partition cannot be altered or deleted.
• Consumers in Kafka work on a pull-based model, continuously polling for messages from their assigned partition. The consumer is responsible for telling Kafka that a message has been processed, and it does so by committing its offset.
  • A consumer can process many messages and accumulate a batch of work before committing its offset. It does not need to commit the offset of the current message to be able to pull and process the next message in the stream.
  • The consumer knows what the next message in the stream is by keeping track of its offset. An offset is a unique, sequential identifier assigned to each message.
• Use Kafka as a message queue
  • If you have processing that can be done asynchronously. Example, transcoding YouTube videos after an upload.
  • If you need to ensure that messages are processed in order.
  • You want to decouple producer and consumer so they can scale independently.
• User Kafka as a stream
  • If you require continuous and immediate processing of incoming data, treating it as a real-time flow. Example, Ad click aggregation.
  • If messages need to be processed by multiple consumers simultaneously. Example, FB Live comments.
• Some numbers:
  • Keep messages under 1 MB for optimal performance.
  • On good hardware, a single broker can store around 1TB of data and handle as many as 1M messages per second.
  • Simplest way to scale: (1) add more brokers (scale horizontally) (2) decide how to partition your data across the brokers.
• To handle hot partitions:
  • Use a compound key: instead of using just the one ID, use a combination of that ID and another attribute, such as geographical region or user ID segments, to form a compound key.
  • Backpressure: slow down the producer. The producer can check the lag on the partition and slow down if it's too high.

Isolation Levels and Locking Mechanisms

In PostgreSQL, isolation levels define the degree to which transactions are separated from the effects of other concurrent transactions in a database system. There are three internal isloation levels: read commited (the default), repeatable read, and serializable.

With serializable isolation level, transactions behave as if they were executed one after another in sequence, preventing all types of concurrency anomalies. The trade off is if two transactions conflict, one will be rolled back and require the application to implement retry logic.

Isolation level sets the broad concurrency policy for a transactions in a db. Row-level locking is used within transactions to explicitly lock specific rows. Row-level locking ensures a consistent view of the data for the duration of the changes to specific rows.

1. PostgreSQL Row-Level Locking:

   • When a row is locked in PostgreSQL (e.g., using SELECT... FOR UPDATE), other transactions attempting to modify that row will wait until the current transaction commits or rolls back.
   • If multiple processes wait for the same row, they will typically queue and acquire the lock in sequence. The db will manage this queue.
   • This mechanism ensures strong consistency but can be a performance bottleneck under heavy contention. It should be held for a short duration on precisely identified rows.

2. Redis-Based Distributed Locks:

   • Redis locks operate on a dedicated Redis instance, separate from the primary application database.
   • They store the state of the lock itself (e.g., if a resource is held), not the primary data.
   • Redis uses atomic operations (like INCR or SETNX) to manage these locks and often includes an expiration time (TTL) to prevent indefinite locks due to crashes.
   • They are used for locking resources across different systems or processes or for longer-term locks, distinct from transactional database locks.

3. When to Use Optimistic Locking:

   • Use when you assume conflicts are rare .
   • Operations are read-only or can be retried.
   • It avoids explicit locks upfront, using a "compare and swap" or versioning mechanism to detect changes before committing.
   • Example: Updating an auction's maximum bid where simultaneous bids are infrequent.
   • Optimistic concurrency is primarily about updating existing data safely.

4. When to Use Row-Level Locking:

   • Use when strong consistency is essential for specific records.
   • You can precisely identify and lock only the necessary rows.
   • The lock can be held for a short duration.
   • Ensures atomic operations within a transaction on specific data.
   • Examples: Updating a user's bank account, processing an auction bid by locking only the auction row, or preventing double-booking of resources.

5. Optimistic vs. Row-Level Locking Relationship:

   • These are generally alternative strategies for handling concurrency, not typically used together for the same operation.

6. Databases Supporting Row-Level Locking:

   • PostgreSQL explicitly supports row-level locking (e.g., SELECT... FOR UPDATE).
   • Cassandra is noted for having "row-level transactions to work with" in certain contexts. Most relational databases generally allow for it.

Two Phase Commit

• This is a protocol that ensures atomocity across multiple participants in a distributed transaction.
• It works by having a coordinator orchestrate the process in two distinct phases: the Prepare Phase (Phase 1) and the Commit Phase (Phase 2).

Redis

• A Redis Cluster is like a library system with multiple branches.
• Node is like one library branch - it's a single Redis instance/server.
• Hash Slots are like catalog sections. Redis divides all possible data into 16,384 numbered sections (0 through 16,383). Each node is responsible for managing some of these sections.
• How data gets distributed:
  • When you store data for user 12345, Redis doesn't directly assign the user to a node. Instead: -Redis takes the key name "user:12345:profile" -Runs it through a math formula to get a number between 0 and 16,383 -That number is the hash slot
    • Whichever node owns that hash slot stores the data

System Failure Modes

Data Loss & Durability

No data lost, ever.

• Failure: Data vanishes due to crashes or incomplete operations.
• Mitigation:
  • Durable Queues: Use Kafka/SQS to buffer and guarantee delivery of critical data.
  • Replication: Keep multiple copies of data across nodes/zones (e.g., database replicas, Kafka replication).
  • WAL/Snapshots: Ensure Write-Ahead Logs and periodic snapshots persist changes for recovery.
  • Event Sourcing/CDC: Capture every state change as an immutable event for audit and replay.

Performance Bottlenecks

System slow or overwhelmed.

• Failure: System is slow, overwhelmed by load, or specific data items are "hot".
• Mitigation:
  • Scale Up (Vertical): Use more powerful hardware (CPU, RAM, SSDs).
  • Scale Out (Horizontal): Add more servers to distribute load.
  • Indexing: Create smart indexes (B-tree, inverted, geospatial) to speed up queries.
  • Caching Layers: Add Redis/Memcached, CDN/Edge caches to store frequent data closer to users.
  • Read Replicas: Distribute read load across database copies.
  • Sharding/Partitioning: Divide data/workload across multiple machines/databases.
  • Hot Key Strategies: For popular items, use cache key fanout (multiple cache entries) for reads or key sharding (split item into sub-keys) for writes.
  • Stale-While-Revalidate: Serve old data while refreshing cache in background to prevent thundering herds.
  • Batching/Queues: Aggregate many small writes into fewer, larger operations or use queues to smooth out write bursts.

Consistency Issues

Data out of sync or wrong.

• Failure: Data appears different across parts of the system or updates conflict.
• Mitigation:
  • CAP Theorem Choice: Prioritize Consistency (e.g., financial systems) or Availability (e.g., social feeds) during network partitions.
  • ACID Transactions: Use database transactions with appropriate isolation levels (e.g., Serializable) for critical operations.
  • Optimistic Concurrency: Assume no conflict, then check and retry if conflict occurs.
  • Distributed Locks: Use Redis/ZooKeeper locks to ensure only one process modifies a resource at a time.
  • Cache Invalidation: Implement Time-To-Live (TTL), write-through, or versioned keys to manage cache freshness.
  • Reconciliation: For external systems, use background jobs to correct eventual inconsistencies.

Service & Component Failures

Parts of the system die

• Failure: Individual servers, services, or jobs crash or become unresponsive.
• Mitigation:
  • Redundancy: Design for no single point of failure through replication and distributed architecture.
  • Failover/Recovery: Automatically promote replicas or restart failed services.
  • Health Checks: Monitor components and re-route traffic from unhealthy ones.
  • Job Retries/DLQs: Use queues with visibility timeouts and Dead Letter Queues to retry failed jobs safely.
  • Pipelining: Break complex jobs into smaller, fault-tolerant stages.

Network & Communication Issues

Things breaking in transit

• Failure: Messages lost, high latency from network, or difficult real-time updates.
• Mitigation:
  • Durable Queues: As mentioned, use Kafka/SQS for reliable message delivery.
  • Real-time Protocols: Choose WebSockets (bi-directional), Server-Sent Events (SSE) (server-to-client push), or Long Polling for efficient updates.
  • Consistent Hashing: Route clients/data to specific servers for stateful connections (e.g., Google Docs, WhatsApp).
  • Pub/Sub: Decouple services and enable broadcasts to many subscribers.
  • Data Locality: Place data and services geographically closer to users (CDNs, regional deployments).
  • Connection Pooling: Reuse network connections to reduce overhead and latency.

Security Vulnerabilities

System compromised

• Failure: Unauthorized access, data breaches, or abuse.
• Mitigation:
  • API Gateway Security: Implement authentication, authorization, rate limiting at the entry point.
  • Request Signing: Use cryptographic signatures (HMAC) to verify request authenticity and prevent replay attacks.
  • Data Protection: Employ encryption (in transit and at rest), tokenization for sensitive data (e.g., credit cards).
  • Rate Limiting/Throttling: Control request volume to prevent abuse and scraping.

Resources

• System Design Handbook
• Hello Interview