Distributed Systems Fundamentals

Introduction and Context

Cloud applications are rarely executed by one computer performing all work in isolation. A typical application may involve web clients, application services, databases, caches, message brokers, object storage, identity services, and external APIs. These components may run on different machines, in different availability zones, or even in different geographic regions. Such an application is a distributed system: a collection of independent computing components that communicate over a network and cooperate to provide a service.

Serverless applications and microservices depend heavily on distributed-systems principles. A serverless function may invoke another function, publish an event, read from a database, and call a third-party API. Each interaction introduces network delay and the possibility that one component may fail while others continue operating. Consequently, cloud developers must reason about concurrency, latency, scalability, availability, consistency, and fault tolerance rather than treating the application as a single reliable program.

A central difficulty is that a distributed system cannot observe the entire system state instantaneously. A service may be healthy from its own perspective while another service is unreachable. Messages may be delayed, duplicated, reordered, or lost. Two replicas may temporarily contain different values. Good architecture therefore does not attempt to eliminate all uncertainty; it detects, contains, and recovers from it.

Learning Objectives

By the end of this chapter, you should be able to:

1. Explain distribution, concurrency, communication, latency, and partial failure.
2. Distinguish scalability, availability, reliability, and fault tolerance.
3. Explain consistency models, including strong and eventual consistency.
4. Apply the CAP theorem to realistic cloud-system design decisions.
5. Select practical techniques for designing resilient microservices and serverless applications.
6. Analyse trade-offs using a distributed cloud case study.

1. The Nature of Distribution

1.1 Independent components and network communication

In a local program, a function call generally occurs within one process and uses shared memory. In a distributed system, a service call usually requires serialising data, transmitting it through a network, waiting for a response, and interpreting possible errors. The remote operation can fail even when the receiving service is functioning correctly.

Common distributed components include:

• Clients: browsers, mobile applications, devices, or other systems.
• Compute services: virtual machines, containers, functions, and managed application platforms.
• Data services: relational databases, NoSQL databases, caches, and object stores.
• Messaging services: queues, topics, event buses, and streaming platforms.
• Control services: authentication, configuration, service discovery, and monitoring.

Distribution provides benefits. Components can be deployed independently, scaled separately, and owned by different teams. However, it also creates coordination costs, operational complexity, and failure modes that do not exist in a single-process application.

1.2 Location transparency is limited

An abstraction may make a remote operation look like a local function call, but the two are not equivalent. A local function usually has predictable execution semantics; a remote call has variable latency, serialization overhead, timeout behaviour, and network failure. Developers should therefore make remote boundaries explicit in architecture and code.

A useful design rule is: assume every network call can be delayed, rejected, duplicated, or interrupted. This does not mean every call will fail, but it means the system must have a defined response when it does.

2. Concurrency and Coordination

2.1 Concurrency and parallelism

Concurrency means that multiple activities are in progress during overlapping periods. Parallelism means that multiple activities execute simultaneously, usually on different processor cores or machines. A system can be concurrent without being parallel; for example, one processor can interleave tasks.

Cloud systems commonly experience concurrency when many users submit requests at the same time, when multiple function instances process events, or when several services update related data. Concurrency is valuable because independent work can progress without waiting unnecessarily. It also creates hazards:

• Two requests may update the same record.
• A message may be processed twice.
• A later event may arrive before an earlier event.
• A retry may overlap the original attempt.

2.2 Race conditions and idempotency

A race condition occurs when the result depends on the timing or ordering of concurrent operations. For example, two workers may read an account balance of 100, each subtract 30, and both write back 70. The correct result should be 40, but the lost update produces 70.

Common controls include transactions, optimistic concurrency control, version numbers, locks, atomic database operations, and partitioning work so that one owner processes a particular key.

An operation is idempotent if applying it more than once has the same intended final effect as applying it once. Setting an order status to “PAID” is usually idempotent; adding 10 to a balance is not inherently idempotent. Idempotency keys are particularly important for payment and order APIs because clients and gateways may retry requests.

A practical pattern is to store a client-generated request identifier with the result. When the same identifier is received again, the service returns the stored result instead of repeating the business operation.

2.3 Ordering

Distributed systems often provide ordering only within a limited scope, such as one queue partition or one entity key. Global ordering across all events is expensive and may reduce scalability. Applications should define the minimum ordering requirement. For example, events for one bank account may require order, while events for unrelated accounts do not.

3. Latency, Throughput, and Timeouts

3.1 Latency and throughput

Latency is the time required to complete an operation, usually measured in milliseconds. Throughput is the amount of work completed per unit of time, such as requests per second or messages per second. A system may have high throughput but poor individual response time, or low latency for small loads but insufficient capacity at scale.

For a sequential request that makes remote calls, a simplified latency model is:

Here, L_total is total response latency; L_client is local client and gateway processing time; L_network is network transmission time; L_i is the processing latency of dependent service i; and L_queue is queueing delay. The model assumes the calls represented by the sum are sequential. If calls are independent and made concurrently, their critical-path contribution is closer to the maximum latency rather than the sum.

For two independent calls made concurrently:

This can reduce latency, but it may increase load and complicate error handling. Parallel calls are appropriate only when their operations are independent and the application can handle partial results or failure.

3.2 Tail latency

Average latency can hide unacceptable user experiences. The 95th percentile latency, written p95, is the value below which 95% of observations fall. The 99th percentile, p99, describes a more extreme tail. If a request depends on many services, a slow tail in any dependency can affect the end-to-end tail. Monitoring should therefore include percentiles, not only averages.

3.3 Timeouts and retries

A timeout prevents a caller from waiting indefinitely. A retry can recover from a temporary fault, but retries also create extra traffic. If many clients retry simultaneously after a dependency slows down, a retry storm may worsen the outage.

Sound retry policies include:

• finite retry limits;
• exponential backoff;
• random jitter to spread retries over time;
• retrying only errors likely to be temporary;
• idempotency protection;
• circuit breakers or admission control.

A timeout should reflect the business operation and the end-to-end latency budget. Setting an arbitrarily long timeout can exhaust threads, connections, or function concurrency.

4. Partial Failure and Fault Tolerance

4.1 Partial failure

In a single computer, a process may fail and the operating system can often identify that failure. In a distributed system, uncertainty is more difficult. A caller that receives no response cannot immediately know whether the remote service failed, the request was lost, the response was lost, or the service completed the operation but became unreachable afterward.

This condition is called partial failure: some components continue functioning while others fail or become unreachable. Partial failure is a defining characteristic of distributed systems.

4.2 Failure classes

Most ordinary cloud applications assume crash-stop or omission-style failures rather than malicious behaviour. Security-sensitive systems may also need to consider arbitrary or Byzantine behaviour, but that requires more specialised protocols.

4.3 Fault tolerance and resilience

Fault tolerance is the capability to continue providing an acceptable service despite specified faults. Resilience is broader: it includes resisting, recovering from, and adapting to disruption.

Important techniques include redundancy, replication, health checks, load balancing, queues, dead-letter queues, circuit breakers, bulkheads, graceful degradation, backups, and automated recovery. These techniques do not make failure impossible; they reduce its effect and shorten recovery time.

A bulkhead isolates resources so that one failing workload does not consume all connections, threads, memory, or function concurrency. A circuit breaker temporarily stops calls to a failing dependency, allowing it to recover and preventing callers from wasting resources.

5. Scalability and Availability

5.1 Scalability

Scalability is the ability of a system to handle increasing workload by adding resources or changing its design. Vertical scaling increases the capacity of an individual machine. Horizontal scaling adds machines or instances. Cloud-native systems generally favour horizontal scaling, although databases and specialised workloads may use both.

A simple capacity approximation is:

R_capacity is estimated system capacity in requests per second, N is the number of equivalent instances, and r_instance is sustainable capacity per instance. This estimate assumes instances are sufficiently independent and that shared bottlenecks such as databases, network links, and rate limits are not the limiting factors. Real capacity is often lower because of coordination and contention.

Scaling dimensions may include requests per second, concurrent users, data volume, geographic coverage, and event-processing rate. Stateless services are easier to scale horizontally because any instance can handle a request. State must then be stored in a shared or partitioned data service rather than local ephemeral memory.

5.2 Availability and reliability

Availability describes whether a service is operational and accessible when needed. A simplified availability ratio is:

A is availability, U is uptime, and D is downtime over a stated measurement period. Both U and D must use the same time unit. Availability targets should specify the service boundary, measurement window, and exclusions.

If independent components must all be available for a request, a simplified series model is:

This assumes component failures are independent and that the request cannot succeed when any listed component is unavailable. Real cloud failures are often correlated, so this equation is an approximation.

Redundancy can improve availability. For two independent replicas where either one can serve a request:

The formula assumes independent failures and correct failover. It does not account for shared network, power, software, or operational failures.

Reliability usually concerns the probability of performing without failure for a specified period. A service can be highly available while still experiencing frequent short failures if recovery is rapid. Availability and reliability are related but not interchangeable.

6. Consistency and Replication

6.1 Why replication is used

Replication stores data or service capability in multiple locations. It can improve read performance, geographic access, and availability. It also creates a consistency problem: replicas may not receive updates simultaneously.

A write may first reach a primary replica and later be copied to other replicas. During that interval, readers can observe different values. Replication therefore requires an explicit consistency model.

6.2 Consistency models

Strong consistency means that a read reflects the latest completed write according to the system’s defined ordering. It simplifies some applications but can require coordination and may increase latency or reduce availability during communication problems.

Eventual consistency means that, if updates stop and communication continues, replicas will eventually converge to the same value. It does not guarantee an exact convergence time unless the service specifies one. During convergence, a read may return an older value.

Other useful models include:

• Read-your-writes consistency: a client sees its own successful updates.
• Monotonic reads: a client does not move backward to an older observed version.
• Causal consistency: causally related operations are observed in an order that respects their cause-and-effect relationship.
• Session consistency: guarantees such as read-your-writes are applied within a client session.

Eventual consistency is suitable for product catalogues, social feeds, analytics dashboards, and replicated content where temporary staleness is acceptable. Stronger guarantees are generally required for inventory reservation, account balances, access-control decisions, and unique identifiers.

6.3 Designing for eventual consistency

Applications should make staleness visible and manageable. Techniques include version numbers, timestamps, conflict-resolution rules, reconciliation jobs, read-after-write routing, user-interface messages such as “processing,” and domain-specific state machines.

A common mistake is to treat eventual consistency as “inconsistent data.” Eventual consistency is a defined guarantee, whereas an uncontrolled race condition or lost update is a defect. The application must specify which temporary differences are acceptable and how conflicts are resolved.

7. The CAP Theorem

The CAP theorem concerns a distributed data system during a network partition. Its three properties are:

• Consistency (C): every read receives the latest write or an error, according to the system’s consistency definition.
• Availability (A): every request to a non-failing node receives a non-error response, without guaranteeing the latest value.
• Partition tolerance (P): the system continues operating despite dropped or delayed messages between nodes.

The theorem states that when a partition occurs, a system cannot simultaneously guarantee both strong consistency and availability. In a real distributed system, partitions must generally be tolerated; therefore, the practical decision during a partition is often between preserving consistency and continuing to serve responses.

CAP is frequently misunderstood. It does not say a system can choose only two properties under normal conditions, nor does it mean “availability” is the same as uptime or performance. It describes a trade-off in the presence of a partition and depends on precise definitions.

For example, a strongly consistent order service may reject writes while its replicas cannot communicate, protecting against double allocation. A highly available social-feed service may accept updates locally and reconcile them later, favouring availability and eventual consistency.

CAP also does not replace other models. Latency, durability, operational cost, security, and recovery objectives remain important. The related PACELC perspective asks what trade-off exists when there is no partition: lower latency or stronger consistency.

8. Practical Architectural Patterns

8.1 Synchronous and asynchronous communication

Synchronous calls are straightforward and appropriate when the caller needs an immediate result. Excessive synchronous chains create long critical paths and amplify failure. Asynchronous communication through queues or events decouples producers and consumers, absorbs bursts, and permits independent recovery. Its costs include delayed results, duplicate delivery, ordering concerns, and more difficult tracing.

8.2 At-least-once delivery and duplicate handling

Many managed messaging systems provide at-least-once delivery: a message is delivered one or more times. Consumers must therefore be idempotent. A common design stores a processed-message identifier in durable storage before or atomically with the business effect. The exact transaction strategy depends on whether the message broker and database support a shared transaction boundary.

8.3 Saga and compensating actions

A distributed transaction across independent services is difficult to coordinate. A saga divides a business transaction into local transactions. If a later step fails, compensating actions semantically undo earlier steps. For example, a booking workflow may reserve a seat, authorise payment, and issue a ticket. If ticket issuance fails, the system may cancel the payment authorisation and release the seat.

Compensation is not always a perfect reversal. Sending an email or changing an external system may be irreversible. The workflow must therefore define states, retries, manual intervention, and audit records.

8.4 Observability

Distributed systems require logs, metrics, and traces. Correlation or trace identifiers should travel across service and message boundaries. Useful indicators include request rate, error rate, latency percentiles, queue depth, retry count, age of oldest message, saturation, and replica lag.

Observability is also a governance concern. Logs should avoid unnecessary personal data, apply retention rules, restrict access, and support investigation without exposing secrets.

9. Applied Case: Illustrative Online Grocery Order Platform

Background

Consider an illustrative online grocery platform built from an API gateway, serverless order functions, an inventory service, a payment provider, a queue, a notification service, and a replicated product catalogue. The platform experiences large evening demand peaks and must continue accepting browsing requests even when a non-critical service is degraded.

Problem

The original design performs inventory reservation, payment, email notification, and order confirmation in one synchronous request. During peaks, payment latency causes request timeouts. Clients retry, producing duplicate orders. A replicated catalogue occasionally displays an old price, while two customers can attempt to purchase the last unit simultaneously.

Analysis

The design has several distributed-systems weaknesses:

1. A long synchronous dependency chain increases latency and creates multiple failure points.
2. Retries are not protected by idempotency keys.
3. Inventory updates lack an atomic conditional check.
4. Catalogue replication is eventually consistent but is treated as authoritative for stock.
5. A payment timeout does not reveal whether payment succeeded.

The business must distinguish acceptable staleness from unacceptable inconsistency. A slightly old product description is acceptable; selling more units than available is not.

Approach and solution

The redesigned workflow uses the following principles:

• The client submits an order with an idempotency key.
• The order service records a PENDING order and publishes an event.
• Inventory performs an atomic conditional reservation: reserve only if available quantity is at least the requested quantity.
• Payment is authorised using the same idempotency key where supported by the provider.
• A saga coordinates order state transitions such as PENDING, RESERVED, PAYMENT_AUTHORISED, CONFIRMED, and CANCELLED.
• Queue consumers use bounded retries and a dead-letter queue.
• Notifications are asynchronous and do not block order confirmation.
• The product catalogue may remain eventually consistent, but authoritative inventory is read from the inventory service.
• Trace identifiers connect the API request, events, function executions, payment calls, and database operations.

The conditional reservation can be expressed conceptually as an atomic operation:

Here, q is the requested quantity. The comparison and update must occur atomically; separate read and write operations would reintroduce a race condition.

Outcome

The revised system reduces the synchronous critical path, prevents duplicate processing, and protects inventory correctness. Some customers may wait briefly for final confirmation, and a payment or notification may be delayed. These are explicit business trade-offs rather than accidental failures. Operators can inspect stuck workflows and replay safe events.

Learning points

The case demonstrates that distributed design begins with business invariants. The system does not need strong consistency everywhere. It needs strong protection for scarce inventory and payment state, while permitting eventual consistency for catalogue views and notifications. It also demonstrates that retries are safe only when operations are idempotent and that asynchronous architecture changes user experience as well as technical behaviour.

10. Limitations, Governance, and Contemporary Practice

Distributed systems introduce cost and complexity. More services mean more deployments, interfaces, credentials, dashboards, alerts, and failure modes. Microservices should not be created merely to increase the number of components. A modular monolith may be preferable when the domain is small, the team is inexperienced, or independent scaling is unnecessary.

Cloud providers offer managed databases, queues, serverless functions, and multi-region replication, but managed services do not remove distributed-systems problems. They move responsibility from hardware operation toward configuration, data modelling, limits, billing, security, and recovery testing.

Governance should address data residency, privacy, encryption, access control, auditability, retention, disaster recovery, and third-party dependency risk. Replicating personal data across regions may conflict with legal or organisational requirements. Automated decisions should be explainable and reviewable where they affect users significantly.

AI-enabled services add further distributed concerns. Model inference may be remote and variable in latency; prompts or documents may contain sensitive information; retries can duplicate costly operations; and model outputs may be nondeterministic. AI components should be treated as dependencies with explicit timeouts, fallback behaviour, privacy controls, and quality monitoring.

Best practices include:

1. Define business invariants and consistency requirements before choosing technology.
2. Make remote calls explicit and assign realistic timeouts.
3. Use idempotency for externally retried commands.
4. Prefer bounded retries with backoff and jitter.
5. Separate critical and non-critical dependencies.
6. Use queues to absorb bursts and isolate failures where appropriate.
7. Test duplicate, delayed, reordered, and missing messages.
8. Measure tail latency, saturation, queue age, and dependency health.
9. Exercise backup restoration and regional-failure procedures.
10. Keep service boundaries understandable and avoid unnecessary distribution.

Chapter Summary

A distributed system consists of independent components that coordinate through networks. Distribution enables independent deployment, elasticity, and geographic reach, but introduces concurrency hazards, variable latency, partial failure, and replica inconsistency. Cloud developers must distinguish latency from throughput, availability from reliability, and scalability from simple resource increase.

Concurrency requires controls such as atomic updates, versioning, transactions, ordering rules, and idempotency. Timeouts, bounded retries, backoff, circuit breakers, bulkheads, queues, and dead-letter handling provide practical fault tolerance. Replication improves access and resilience but requires an explicit consistency model. Eventual consistency is useful when temporary staleness is acceptable; strong guarantees are needed for critical invariants such as inventory and financial state.

CAP explains the trade-off between strong consistency and availability during a network partition. It does not eliminate the need to consider latency, durability, security, cost, and operational complexity. Effective serverless and microservice architectures use asynchronous workflows, sagas, observability, and domain-specific consistency rather than applying one guarantee to every component.

Key Terms

Distributed system: A collection of independent components that communicate over a network to provide a combined service.

Concurrency: Overlapping progress of multiple activities.

Parallelism: Simultaneous execution of multiple activities.

Latency: Time taken to complete an operation.

Throughput: Amount of work completed per unit of time.

Tail latency: High-percentile response time, such as p95 or p99 latency.

Partial failure: Failure of some components while others continue operating.

Timeout: A limit on how long a caller waits for an operation.

Idempotency: Property that repeated application of an operation has the same intended final effect as one application.

Race condition: A defect caused by dependence on timing or ordering of concurrent operations.

Scalability: Ability to handle increased workload by adding resources or changing design.

Availability: Proportion of a stated period during which a service is operational and accessible.

Reliability: Probability or capability of operating without failure for a specified period.

Fault tolerance: Continued acceptable operation despite specified faults.

Replication: Maintaining copies of data or capability in multiple locations.

Strong consistency: A guarantee that reads reflect the latest completed write according to defined ordering.

Eventual consistency: Guarantee that replicas converge if updates stop and communication continues.

CAP theorem: The result that consistency and availability cannot both be guaranteed during a network partition while partition tolerance is maintained.

Saga: A sequence of local transactions coordinated with compensating actions.

Observability: Ability to understand internal system behaviour through logs, metrics, and traces.

Review Questions

Short Answer Questions

1. What makes a distributed system different from a single-process application?
2. Distinguish concurrency from parallelism.
3. Why are idempotency keys useful in serverless applications?
4. Define eventual consistency and give one suitable use case.
5. What does a circuit breaker protect against?

Long Answer Questions

1. Explain partial failure and discuss four techniques for designing fault-tolerant cloud services.
2. Compare strong consistency and eventual consistency, including advantages, limitations, and appropriate applications.
3. Explain the CAP theorem and clarify why it applies specifically to network partitions.
4. Discuss how queues, retries, timeouts, and dead-letter queues work together in an event-driven system.
5. Explain the difference between availability, reliability, and scalability, and describe how each should be measured or evaluated.

Analytical / Application Questions

1. An order service calls inventory, payment, and notification services synchronously. Analyse the likely latency and failure problems, then propose a redesigned workflow.
2. Two workers may reserve the last unit of a product at the same time. Explain the race condition and design an atomic or transactional solution.
3. A replicated product catalogue may be several seconds out of date, but the inventory service is authoritative. Decide which operations may use eventual consistency and justify your decision.

Multiple Choice Questions with Answers

Question 1

Which condition is characteristic of a partial failure?

A. Every component stops at exactly the same time

B. Some components fail while others continue operating

C. A program completes without using a network

D. All replicas always contain identical data

Correct Answer: B. Some components fail while others continue operating

Question 2

What is the primary purpose of an idempotency key?

A. To increase network bandwidth

B. To guarantee zero latency

C. To prevent a retried request from producing an unintended duplicate effect

D. To encrypt a database table

Correct Answer: C. To prevent a retried request from producing an unintended duplicate effect

Question 3

If two independent remote calls are made concurrently, their ideal critical-path contribution is approximately which expression?

A. L_1 + L_2

B. L_1 × L_2

C. max(L_1, L_2)

D. L_1 ÷ L_2

Correct Answer: C. max(L_1, L_2)

Question 4

Which statement best describes eventual consistency?

A. Every read always returns the newest value

B. Replicas converge if updates stop and communication continues

C. No replica is allowed to accept writes

D. A system cannot recover from a network partition

Correct Answer: B. Replicas converge if updates stop and communication continues

Question 5

During a network partition, what trade-off is identified by the CAP theorem?

A. Security versus encryption

B. Strong consistency versus availability

C. Throughput versus storage capacity only

D. Authentication versus authorisation

Correct Answer: B. Strong consistency versus availability

Question 6

Which technique helps prevent a failing dependency from exhausting caller resources?

A. Circuit breaker

B. Unbounded retries

C. Removing all timeouts

D. Duplicating every request indefinitely

Correct Answer: A. Circuit breaker

Question 7

A message is delivered twice by an at-least-once queue. What should the consumer generally do?

A. Assume duplicates are impossible

B. Process both messages as new commands

C. Use deduplication or an idempotent operation

D. Delete the database

Correct Answer: C. Use deduplication or an idempotent operation

Question 8

Which operation is most naturally idempotent?

A. Add 10 to an account balance

B. Append an item to a list without a unique key

C. Set an order status to PAID

D. Increment a retry counter

Correct Answer: C. Set an order status to PAID

Question 9

What does horizontal scaling usually mean?

A. Adding more equivalent instances

B. Increasing the clock speed of one processor only

C. Reducing the number of users

D. Deleting replicated data

Correct Answer: A. Adding more equivalent instances

Question 10

Why should inventory reservation use an atomic conditional update?

A. To allow two workers to sell the same final unit

B. To prevent a race between checking availability and reducing quantity

C. To make catalogue descriptions strongly consistent everywhere

D. To eliminate the need for monitoring

Correct Answer: B. To prevent a race between checking availability and reducing quantity