[Week 2] Distributed Systems Fundamentals: Consistency, Replication, Sharding, and Queues Explained

The Core Distributed Systems Concepts Every Engineer Should Understand Before Attempting a Senior System Design Interview, From Consistency to Queues

What This Blog Will Cover

• Why distributed systems are hard

• Consistency models explained

• Replication in depth

• Sharding and partitioning

• Queues and async processing

This is the second post in the eight-week From Foundations to FAANG-Ready series. If you missed Week 1, it covered scalability, latency, and load balancing, the infrastructure fundamentals that this week builds directly on.

Each week assumes the previous one. Subscribe to get all eight delivered to your inbox.

Subscribe to System Design Nuggets to unlock the full eight-week series and weekly system design resources.

---

A single machine is a beautiful thing from a correctness standpoint.

When you write data to it, that data is there. When you read it back, you get what you wrote.

When two operations happen at the same time, the machine processes them one after the other. Everything is sequential, everything is visible, and the mental model is simple.

The moment you add a second machine, something fundamental changes. Now data might live in two places, and those two places might disagree.

A write to one machine might not have reached the other yet. Two operations might happen simultaneously on different machines with conflicting results.

The simplicity of the single machine disappears, and a new class of problems appears that has no equivalent in single-machine computing.

These are distributed systems problems, and they are what make scaling hard. Not the raw mechanics of adding servers, but the correctness challenges that arise when data is spread across many machines that communicate over an unreliable network.

The engineers who design systems well at scale are the ones who understand these problems precisely, because you cannot solve a problem you have not clearly defined.

This week covers four foundational distributed systems concepts, each one addressing a different facet of this challenge.

Consistency is about what data readers are allowed to see and when.

Replication is about how copies of data are kept in sync across machines.

Sharding is about how data is split across machines to scale beyond what one can hold. And queues are about how components communicate without creating tight dependencies that break under load.

Together, these four concepts underlie the majority of scaling and reliability decisions in system design.

Part 1: Consistency

Consistency is the most misunderstood concept in distributed systems, and it is also the one that interviewers probe most carefully because shallow answers are easy to spot and deep answers are rare.

Why Consistency Is Hard

In a single machine with one copy of the data, consistency is automatic. A write completes and every subsequent read sees the new value. There is nothing to reason about.

In a distributed system with multiple copies of the data, this guarantee breaks. A write goes to one copy, and before it reaches the others, a reader might read from one of those other copies and see the old value. How long this inconsistency lasts, under what conditions it occurs, and what guarantees the system provides about it, is what consistency models describe.

The reason this matters is that different consistency models have different costs. Stronger consistency requires more coordination between machines, which adds latency and reduces availability.

Weaker consistency allows disagreement between copies, which is cheaper but requires the application to tolerate or handle stale reads. Choosing the right model for each piece of data is one of the most important decisions in distributed system design.

Strong Consistency

Strong consistency, also called linearizability, means every read reflects the most recent write. No matter which copy of the data a reader contacts, they see the latest value. The system behaves as if there is only one copy, even though there are many.

Achieving strong consistency requires coordination. Before a write is acknowledged, it must be confirmed on enough copies that any subsequent read is guaranteed to find it, regardless of which copy is consulted. This coordination takes time, adding latency to every write, and if some copies are unavailable the write might not complete at all.

Strong consistency is the right choice when the data must always be correct from the reader’s perspective.

An account balance, an inventory count for a high-demand item, or a seat reservation for a flight where double-booking is catastrophic all require strong consistency. The cost in latency and availability is worth paying because the cost of a wrong answer is higher.

Eventual Consistency

Eventual consistency means that copies will converge to the same value over time, but at any given moment different copies might return different values. A write to one copy will eventually propagate to the others, but reads during the propagation window might see the old value.

Eventual consistency is cheaper than strong consistency because it requires no coordination on writes. A write goes to one copy and is acknowledged immediately, without waiting for other copies to confirm. The propagation happens in the background asynchronously.

Eventual consistency is the right choice when brief staleness is acceptable.

A like count on a social post, a view count on a video, or a follower count on a profile all tolerate being slightly out of date.

If the like count reads as 47 for a few hundred milliseconds after the 48th like is added, nobody is harmed and the system is cheaper to run. Applying strong consistency here wastes latency and availability for a correctness guarantee the application does not need.

The practical skill is not knowing these definitions but applying them. Different data within the same system warrants different consistency levels.

A payment record needs strong consistency. The list of recommended products below a checkout page is fine with eventual consistency. Making this distinction explicitly is what good distributed system design looks like.

Read-Your-Writes Consistency

Between strong and eventual consistency sits a practically important middle ground called read-your-writes consistency. This guarantees that after a user writes something, their own subsequent reads will reflect that write, even if other users might still see the old value.

This matters because the most jarring consistency violation from a user’s perspective is when they make a change and then immediately do not see it. A user who posts a comment and then refreshes the page to find their comment missing has a worse experience than one who posts a comment and sees it instantly while someone else takes a few seconds longer to see it.

Read-your-writes is achieved by routing each user’s reads to the copy that handled their write, or by tracking the position of their last write and only serving reads from copies that have caught up to at least that position. It is a common consistency requirement for any system where users interact with their own data.

Monotonic Reads

Another practically important consistency property is monotonic reads, which guarantees that if a user reads a value, subsequent reads will never return an older value. Without this guarantee, a user might see a value change and then appear to revert, as if time is going backward.

This happens when reads are distributed across replicas with different replication lag. Read one goes to a replica that has caught up and returns the new value. Read two goes to a replica that has not caught up and returns the old value. The user sees the value revert, which is disorienting even if the reversion is brief.

Monotonic reads is achieved by sticking a user to the same replica for the duration of a session, or by always reading from copies that have caught up past a certain watermark.

The CAP Theorem: What It Actually Means

The CAP theorem is cited constantly and understood rarely. It states that a distributed system can guarantee at most two of three properties: Consistency, Availability, and Partition Tolerance.

The crucial insight is that partition tolerance is not optional.

A network partition, where some machines cannot communicate with others, is an inevitable event in any real distributed system. Networks fail, switches crash, cables get cut.

A system that cannot tolerate partitions is a system that will fail in production. So partition tolerance must be chosen, which means the real choice is between consistency and availability when a partition occurs.

A CP system (Consistency and Partition Tolerance) refuses to serve requests from isolated nodes during a partition rather than risk returning stale or conflicting data. This preserves correctness at the cost of availability.

An AP system (Availability and Partition Tolerance) continues serving requests from all nodes during a partition, accepting that different nodes might return different values temporarily. This preserves availability at the cost of consistency.

The shallow answer to a CAP question is to label systems as CP or AP.

The deep answer recognizes that most systems tune this behavior per operation, that the trade-off only applies during a partition (not during normal operation), and that the right choice depends on what staleness and inconsistency would cost for a specific piece of data.

Saying this out loud in a system design interview is what separates a surface understanding from a real one.

Part 2: Replication

Replication is the practice of keeping copies of data on multiple machines. It serves two purposes: redundancy, so the failure of one machine does not lose data, and scaling, so multiple machines can serve the read traffic.

Leader-Follower Replication

The most common replication pattern is leader-follower, also called primary-replica or master-slave.

One machine, the leader, handles all writes.

One or more machines, the followers, hold copies of the data and can serve reads.

When a write arrives, the leader processes it, records it in its write-ahead log, and replicates the change to the followers. The followers apply the changes to their own copies. From that point, reads can go to either the leader or any follower, and the read is spread across all machines.

Synchronous replication means the leader waits for at least one follower to confirm the write before acknowledging it to the caller. This guarantees that if the leader fails immediately after acknowledging the write, at least one follower has the data. The cost is that the write latency includes the round trip to a follower.

Asynchronous replication means the leader acknowledges the write immediately without waiting for followers. This is faster but creates a window where the leader has acknowledged a write that no follower has yet received. If the leader fails in this window, the write is lost even though it was acknowledged.

Most production systems use asynchronous replication with one follower configured for synchronous replication, balancing the durability guarantee against the latency cost.

Replication Lag and Its Consequences

Replication lag is the delay between a write being committed on the leader and appearing on the followers.

Under normal conditions, this is small, typically milliseconds.

Under load, it can grow to seconds. Understanding what this means for the application is essential.

During the replication lag window, reads from followers return stale data.

A user who writes and immediately reads back might see the old value if their read goes to a lagging follower. This is the read-your-writes problem described earlier.

A user who reads twice might see a newer value and then an older one if their reads go to followers with different lag. This is the monotonic reads problem.

The solution for read-after-write is to route the user’s reads to the leader for a short period after a write, or to route them to the specific follower that handled the write if synchronous replication was used.

Monitoring replication lag and alerting when it exceeds a threshold is important because lag that grows without bound eventually causes followers to be significantly out of date, defeating the purpose of replication.

Handling Leader Failure

When the leader fails, one of the followers must be promoted to become the new leader. This process is called failover, and it must happen automatically for the system to recover without manual intervention.

Automatic failover requires three things: detecting that the leader has failed, choosing which follower to promote, and reconfiguring the system so writes go to the new leader. The detection is done through health checks and timeouts.

The choice is usually the follower with the most up-to-date data, to minimize how much data the new leader is missing. The reconfiguration updates the load balancer or the client routing to point at the new leader.

The hard problem in failover is avoiding split-brain, where the old leader recovers and both it and the new leader accept writes independently, creating conflicting copies of the data.

Preventing split-brain requires that the old leader be definitively fenced out before the new leader begins accepting writes, often by requiring the new leader to obtain a majority quorum from the cluster before proceeding.

Multi-Leader Replication

Multi-leader replication allows writes to go to any of several leaders, each of which replicates to the others. This is used in multi-region systems where routing all writes to one region would add latency for users in other regions.

The price of multi-leader replication is write conflicts.

If two users on different leaders write the same data at the same time, both writes are valid on their respective leaders but they conflict when replicated.

The system must resolve the conflict, either by keeping the last write based on timestamp, which risks silently discarding data because distributed clocks cannot be trusted, by keeping both versions and asking the application to resolve the conflict, or by designing the data to avoid conflicts, for example using data types that can be merged without conflict regardless of order.

Multi-leader replication is powerful for global systems but the conflict resolution problem is genuinely hard and should not be underestimated.

Part 3: Sharding

Replication distributes reads but does not help with writes or storage, since all writes still go to one leader and all leaders hold all the data.

When write volume or data size exceeds what one machine can handle, sharding distributes the data itself.

What Sharding Does

Sharding, also called partitioning, splits data across multiple machines so each holds only a portion. A shard key determines which machine stores each piece of data.

With ten shards, each shard holds approximately one tenth of the data and handles approximately one tenth of the writes, multiplying both write capacity and storage capacity by the number of shards.

The total capacity of the system scales with the number of shards, and adding shards increases capacity without limit in principle. This is what makes sharding the solution for truly large-scale systems.

Choosing the Shard Key

The shard key decision is the most consequential choice in sharding, and getting it wrong causes problems that are expensive to fix because changing the shard key requires redistributing all the data.

A good shard key distributes data and traffic evenly across all shards so no single shard is overloaded while others are idle.

A poor shard key creates hot shards where one or a few shards receive most of the traffic while the rest sit underutilized. A hot shard becomes the bottleneck for the entire system.

Common sources of hot shards include using a timestamp as the shard key for time-series data, since all recent writes go to the current time range’s shard, and using a user ID for data about highly active users, since a celebrity or viral account generates far more traffic than a typical user.

The solution to hot shards is choosing a shard key that distributes uniformly.

A hash of the primary key distributes data evenly regardless of access patterns. Geographic region distributes data by location but creates hot shards for popular regions. User ID distributes evenly in expectation but requires special handling for naturally hot users.

The Cross-Shard Query Problem

Sharding comes with a fundamental limitation: queries that need data from multiple shards must contact each relevant shard, collect the results, and merge them in the application or a query coordinator. This is expensive in both latency and compute.

A join between two sharded tables where the join key is not the shard key requires all combinations of shards to be contacted.

A sort and paginate query across a sharded collection requires each shard to sort and paginate independently, then the results to be merged and re-paginated at the application level.

An aggregate like a count or a sum requires summing the result from each shard.

The standard solution is to design the sharding so that the data a query needs is almost always on the same shard. For a user-centric system, sharding by user ID means all of a user’s data is on one shard and queries about a single user never cross shards.

The queries that do cross shards, like reporting queries that aggregate across all users, are either accepted as slow operations or handled by a separate analytics system.

Consistent Hashing for Dynamic Sharding

When the number of shards changes, a naive approach to sharding requires remapping a large fraction of keys since the hash modulo changes. This means moving enormous amounts of data across the cluster just to add or remove capacity.

Consistent hashing solves this by mapping both data keys and shards onto a hash ring, where each shard is responsible for the range of the ring between it and the previous shard.

When a shard is added, it takes responsibility for a portion of an adjacent shard’s range, and only the keys in that portion need to move.

When a shard is removed, its range is absorbed by an adjacent shard. The total data movement is proportional to the size of the affected range rather than the size of the entire dataset.

Virtual nodes, where each physical shard occupies multiple positions on the ring, improve distribution uniformity and allow gradual capacity changes. This is how Cassandra and DynamoDB distribute data across their clusters.

Part 4: Message Queues and Asynchronous Processing

The first three parts of this week covered how data is kept correct and how it scales. This part covers how components communicate, specifically how to handle the cases where synchronous communication breaks down.

The Problem With Synchronous Communication

Synchronous communication means one component calls another and waits for a response. This is the simplest model and works well when the callee is fast, reliable, and can keep up with the caller.

It breaks down in three situations.

• First, when the callee is slow, the caller is blocked waiting, holding resources and increasing latency for the user.

• Second, when the callee is unavailable, the call fails and the caller must handle the error, often losing the work.

• Third, when the caller produces work faster than the callee can process it, the callee is overwhelmed and the system degrades under load.

Message queues address all three by introducing a buffer between the producer and the consumer.

The producer writes a message to the queue and continues immediately without waiting.

The consumer reads from the queue and processes messages at its own pace. The queue holds messages that have not yet been processed, absorbing the gap between production and consumption.

How Message Queues Work

A message queue holds messages durably until they are successfully consumed.

When a consumer pulls a message, the queue marks it as in-flight so no other consumer receives the same message. When the consumer acknowledges successful processing, the queue deletes the message.

If the consumer fails before acknowledging, the queue makes the message available again after a timeout.

This mechanism provides at-least-once delivery: a message is never lost because it stays in the queue until acknowledged, but it may be delivered more than once if a consumer fails after processing but before acknowledging.

The standard solution is idempotent consumers, designed so processing the same message twice has the same effect as processing it once, using a unique identifier per message to detect and skip duplicates.

A dead-letter queue holds messages that have failed to process after a configured number of attempts. Rather than retrying indefinitely and blocking the queue, messages that cannot be processed are moved aside for inspection. Monitoring the dead-letter queue is important for catching bugs and data quality issues that prevent processing.

Decoupling and Its Benefits

The most important benefit of message queues is not performance but decoupling. When two components communicate through a queue, neither knows about the other.

The producer does not know which consumer will handle its message or how long it will take.

The consumer does not know which producer sent the message or how many more are coming. They are connected by the queue, not by a direct call.

This decoupling has real architectural benefits.

The producer and consumer can be deployed, scaled, and modified independently. Adding a new consumer that processes the same messages for a different purpose requires no changes to the producer. Scaling the consumer to handle more volume requires no changes to the producer.

The consumer can be taken offline for maintenance without the producer knowing or caring.

Decoupling also provides resilience.

If the consumer is slow, messages accumulate in the queue and are processed when capacity is available.

If the consumer crashes, messages wait in the queue until the consumer recovers. The producer keeps working throughout, unaffected by the consumer’s problems.

Publish-Subscribe for Fan-Out

The basic queue model delivers each message to exactly one consumer. Publish-subscribe, or pub-sub, extends this by delivering each message to all subscribers of a topic.

A producer publishes messages to a topic. Multiple consumers subscribe to the topic and each receives a copy of every message. This enables fan-out, where one event needs to trigger multiple independent downstream actions.

An order placed event might need to trigger inventory reservation, notification sending, analytics recording, and fraud checking simultaneously.

With direct calls, the order service must call each downstream service, coupling them all together.

With pub-sub, the order service publishes one event and each downstream service subscribes independently. Adding a new downstream service requires no changes to the order service.

When to Use Async Processing

The decision of when to use asynchronous processing through a queue versus synchronous direct calls is one of the most practical system design skills.

Use asynchronous processing when the work does not need to complete before the response goes back to the user.

Sending a confirmation email, processing an uploaded video, generating a report, and sending a push notification are all operations that can happen after the user receives a response. Making users wait for these operations adds latency for no benefit.

Use asynchronous processing when the downstream service might be slow or unavailable.

A payment notification that would block the checkout experience if the notification service is slow should be queued so checkout is always fast.

Use asynchronous processing when the volume of work is bursty.

A queue absorbs spikes by accepting messages at the rate they arrive and processing them at the sustainable rate the consumer can maintain.

Use synchronous calls when the result is needed before the response can be formed.

A user who submits a payment needs to know whether it succeeded before the checkout completes.

A user who queries their balance needs the current value. These cannot be made asynchronous without fundamentally changing the user experience.

How These Four Concepts Connect

These four concepts are not independent topics. They form an integrated set of tools for the same underlying challenge, which is how to make data correct, available, and efficiently processed when it is distributed across many machines.

Consistency determines what guarantees readers get about the data they see.

Replication makes consistency possible across multiple machines while adding redundancy.

Sharding scales the capacity that replication alone cannot address.

Queues handle the communication between the components that read and write the data.

In a real system, all four appear together.

The database is sharded for scale, with each shard replicated for redundancy and availability, with a consistency level chosen per data type based on what staleness would cost, and with high-volume write paths going through a queue so spikes do not overwhelm the database directly.

Understanding how they fit together is what allows a designer to make coherent decisions rather than applying each pattern in isolation.

A sharding strategy that ignores replication will have hot shards with no redundancy.

A replication strategy that ignores consistency will have correctness bugs under concurrent load.

A queue strategy that ignores idempotency will produce duplicates when consumers fail and retry. The concepts are most useful when applied as a system rather than a checklist.

Key Takeaways

• Consistency describes what readers are allowed to see, from strong consistency where every read sees the latest write, to eventual consistency where copies converge over time, with practically important middle grounds like read-your-writes and monotonic reads addressing specific user-facing correctness requirements.

• The CAP theorem’s real message is that partition tolerance is mandatory and the actual choice is between consistency and availability during a partition, not a static label applied to a system.

• Leader-follower replication scales reads and adds redundancy, with replication lag being the central challenge and failover being the mechanism that handles leader failure without losing data.

• Sharding scales writes and storage by splitting data across machines, with the shard key choice being the most consequential decision and cross-shard queries being the primary limitation.

• Message queues decouple producers from consumers, providing at-least-once delivery through acknowledgment-based message retention, enabling fan-out through pub-sub, and handling bursty traffic through buffering.

• The four concepts work as a system: consistency governs correctness, replication provides redundancy and read scale, sharding provides write and storage scale, and queues handle the communication between all the components.

• Knowing these concepts is only half the value: the other half is applying them to specific data types and access patterns based on what correctness violations and availability trade-offs would actually cost in the specific system being designed.

Next week in Week 3, we cover API design for system design interviews: REST, GraphQL, gRPC, and how to choose between them, how to design clean interfaces, and how to handle the API concerns that interviewers probe most heavily.

If you have not already subscribed, do so now so Week 3 lands in your inbox when it publishes.

The series builds week by week and each post assumes the ones before it.