3.3 Distributed systems¶

Overview and motivation¶

A distributed system is any system whose components run on more than one machine and coordinate over a network. The moment you cross a process boundary over a network, you inherit some hard truths that simply do not exist inside a single process. The network is unreliable, and its latency varies. Messages can be lost, duplicated, delayed, or reordered. Remote components fail on their own. There is no shared clock. The classic "fallacies of distributed computing" (the network is reliable, latency is zero, bandwidth is infinite, the topology never changes) name exactly the assumptions that cause outages. Your job is to design for these realities from the start, rather than rediscover them during an incident.

For a large organization, distribution is not optional. Any system that serves national or global scale, joins up multiple departments, or needs high availability will span many machines, data centers, and often regions. Enterprises run distributed transaction systems, event pipelines, and multi-region deployments. Governments run inter-agency integrations where each agency owns its own systems and no one controls the whole. Here the gap between a robust design and a fragile one shows up as headline outages, missed benefit payments, and regulatory consequences. The techniques in this chapter (consistency reasoning, idempotency, retries with backoff, circuit breakers, sagas, and distributed observability) are your standard defenses.

The hardest part of distributed systems is that failures are partial and intermittent. A single-machine program either works or crashes. A distributed system can be half-working: some requests succeeding, some timing out, and some silently lost, all at once. This chapter focuses on the reasoning and patterns that let a large team build systems that degrade gracefully and stay understandable under partial failure.

Key principles¶

• The network is not reliable. Design every remote interaction assuming it can be slow, fail, duplicate, or reorder.
• You cannot have perfect consistency and perfect availability during a partition. Choose deliberately per interaction (CAP/PACELC), and remember latency is a cost even when there is no partition.
• Make operations idempotent. If an operation can be safely retried, most distributed failure handling becomes tractable.
• Every remote call needs a timeout. Unbounded waits turn one slow dependency into a system-wide outage.
• Prefer eventual consistency where the business allows it, but make it explicit. Users and auditors must understand when they might see stale data.
• Isolate failures. Bulkheads and circuit breakers stop one failing component from cascading into all of them.
• You cannot debug what you cannot see. Distributed flows require correlated tracing, metrics, and logs across every hop.
• "Exactly-once delivery" is a myth; exactly-once processing is an engineering achievement. Design for at-least-once with deduplication.

Recommendations¶

Reason about consistency with CAP and PACELC¶

The CAP theorem says that during a network partition a system must choose between consistency (every read sees the latest write) and availability (every request gets a response). PACELC adds a second trade: Else (when there is no partition) you still trade Latency against Consistency. Do not stamp this as a whole-system label. Decide it per operation. A bank's balance transfer needs strong consistency and will refuse rather than risk a double-spend. A social feed or a product-view counter can accept staleness in exchange for availability and speed. Write down which consistency model each data flow uses (strong, causal, read-your-writes, or eventual) so no one assumes a guarantee the system does not actually provide.

Build idempotency, timeouts, retries, and backoff together¶

Treat these four techniques as one package. Give every remote operation a timeout, so a hung dependency cannot block a thread forever. On failure, retry, but only for operations that are safe to repeat. Safe to repeat means idempotent: assign each request a unique key and have the receiver deduplicate, so a retried "charge card" does not charge twice. Space your retries with exponential backoff and jitter, so you avoid a synchronized retry storm that turns a brief blip into a self-inflicted denial of service. Cap the number of retries and the total time budget, because retrying forever just moves the failure. Without idempotency, retries are dangerous. Without backoff, retries are destructive.

Add circuit breakers and bulkheads to stop cascades¶

A circuit breaker watches calls to a dependency and, after a threshold of failures, "opens": it fails fast for a cooldown period instead of piling more requests onto a struggling service, then "half-opens" to test recovery. This stops the cascade where a slow downstream service exhausts every caller's threads until the whole system stalls. Bulkheads partition resources (thread pools, connection pools) so that saturation in one dependency cannot eat the capacity others need. Pair both with graceful degradation: when a non-critical dependency is unavailable, return cached or default responses rather than failing the whole request.

Manage distributed transactions with sagas, not two-phase commit¶

You usually cannot hold a single ACID (Atomicity, Consistency, Isolation, Durability) transaction across multiple services or databases. Distributed two-phase commit is slow, locks resources, and cuts availability. Use the saga pattern instead. Model a business transaction as a sequence of local transactions, each publishing an event that triggers the next, with a compensating action for each step to undo it if a later step fails. Sagas come in two flavors. Choreography has services react to each other's events, with no central controller. Orchestration has a central coordinator drive the steps, which is easier to reason about and monitor. Sagas embrace eventual consistency: the system passes through intermediate states and then converges. So design the user experience and the audit trail to account for "in-progress" and "compensated" states.

Treat exactly-once as at-least-once plus deduplication¶

Message brokers cannot truly guarantee exactly-once delivery across failures. What they and you can achieve is at-least-once delivery with idempotent processing, which yields exactly-once effects. Design consumers to handle duplicate messages safely, using idempotency keys or a processed-message log. Know your broker's ordering and delivery guarantees precisely. For streaming, use consumer groups, partitions, and offset management on purpose, and make reprocessing safe, so you can replay a stream after a bug fix without corrupting downstream state.

Instrument distributed flows end to end¶

Adopt the three pillars of observability, correlated across service boundaries. Propagate a trace/correlation ID through every hop, so you can follow a single user request across all the services it touches (distributed tracing). Emit structured metrics (latency percentiles, error rates, saturation, throughput) per service and per dependency. Emit structured logs that carry the correlation ID. Use all this to set service-level objectives and to alert on symptoms users actually feel, such as error rate and latency, rather than only on individual machine health. In a distributed system, observability is not optional tooling. It is the only way to understand behavior under partial failure.

Trade-offs: pros and cons¶

The master trade-off is between coordination and independence. Every guarantee you want across machines (consistency, ordering, exactly-once) costs latency, availability, or complexity. It requires machines to agree, and agreement over an unreliable network is expensive. The skill is to buy only the guarantees the business truly needs, operation by operation, and to design everything else for graceful degradation. Over-buy consistency and your systems turn slow and fragile. Under-buy it and you get silent data corruption that surfaces as an audit failure months later.

Examples¶

Startup. A small fintech startup has just two moving parts that talk over the network: its app and a third-party payment provider. Even at this size it makes every charge request carry an idempotency key and wraps the call in a retry with backoff, so a dropped response on a flaky connection never double-charges a customer. Skipping this feels cheap on day one, but the first duplicate charge that hits a real user costs a support fire, a refund, and a dent in trust the young company cannot spare.

Enterprise. A global ride-sharing platform processes trip payments through a saga: authorize card, debit rider, credit driver, record ledger entry, each a local transaction with a compensating reversal. Every step carries an idempotency key, so retries after a network timeout never double-charge. Calls to the fraud-scoring service sit behind a circuit breaker; when it degrades during peak, the breaker opens and trips fall back to a conservative score instead of blocking every ride. When a customer disputes a trip, distributed tracing lets engineers follow it across a dozen services in seconds.

Government. A national identity service is used by many agencies for verification. It offers a strongly consistent read for authoritative status checks (you must not approve a benefit against stale identity data), plus an eventually consistent, cached endpoint for high-volume, non-critical lookups. Inter-agency data exchange runs over a durable message queue with at-least-once delivery, and each agency's consumer deduplicates on a message ID, so a redelivered record does not create a duplicate case. Correlation IDs flow across agency boundaries, giving auditors an end-to-end trace of how a citizen's data moved between departments.

Business case: motivations, ROI, and TCO¶

Distributed-systems discipline is bought cheaply and its absence is paid for catastrophically. The adoption cost is engineering time to build idempotency, timeouts, retries, circuit breakers, and tracing into shared libraries and platform defaults. That is a modest, mostly one-time investment that then benefits every team. The cost of not adopting it is measured in major outages: a single missing timeout that cascades into a full platform outage, a non-idempotent payment path that double-charges thousands of customers, or a saga-less distributed transaction that leaves data permanently inconsistent. Each of these is a headline incident with direct revenue, remediation, and reputational cost, and in regulated sectors, fines.

Frame the case to leadership around availability and blast radius. Resilience patterns directly reduce both the frequency and the duration of severe incidents, the metrics executives already track as uptime and mean-time-to-recovery. Distributed observability is the single largest lever on MTTR: teams with correlated tracing resolve cross-service incidents in a fraction of the time. Because these capabilities are best delivered as shared platform defaults, their per-team marginal cost is low and their organization-wide payoff compounds. The TCO argument is simple: building resilience in from the start is a fraction of the cost of retrofitting it after the outage that forces the issue.

Anti-patterns and pitfalls¶

• No timeouts. A single hung dependency exhausts every thread and takes down the whole system.
• Retrying non-idempotent operations. Duplicate side effects: double charges, duplicate records, doubled emails.
• Retry storms. Synchronized retries without backoff and jitter that amplify a small blip into an outage.
• Assuming exactly-once delivery. Building consumers that break on duplicate messages the broker will eventually deliver.
• Distributed transactions via shared database. Coupling services through one database to fake ACID, recreating a distributed monolith.
• Ignoring partial failure. Code that assumes a remote call either fully succeeds or fully fails, with no handling for "timed out but maybe completed."
• No correlation IDs. Debugging a cross-service incident by grepping unrelated logs on ten machines.
• Chatty synchronous call chains. Deep synchronous dependency graphs where any one slow hop stalls the entire request.

Maturity model¶

• Level 1: Initial. Remote calls treated like local calls. No timeouts or retries, or naive retries. Failures cascade. Debugging is per-machine log spelunking.
• Level 2: Managed. Timeouts and basic retries exist but are inconsistent. Some idempotency. Logs are centralized but not correlated. Distributed transactions are hoped to work.
• Level 3: Defined. Idempotency, backoff, circuit breakers, and bulkheads are standard via shared libraries. Sagas handle multi-service transactions. Distributed tracing with correlation IDs is in place. Consistency models are documented per flow.
• Level 4: Optimizing. Resilience is the platform default and continuously tested (including fault injection). SLOs drive alerting on user-visible symptoms. Teams reason explicitly about consistency and delivery guarantees, and the system degrades gracefully by design.

Ideas for discussion¶

1. Which of your critical operations are genuinely idempotent today, and which quietly are not?
2. For each major data flow, can your team state the consistency model and the delivery guarantee from memory?
3. Where would a circuit breaker have prevented your last cascading outage?
4. How long does it currently take to trace a single failing request across all the services it touches?
5. Which of your "distributed transactions" are actually relying on luck, and which are true sagas with compensations?
6. If your message broker redelivered every message twice for an hour, what would break?

Key takeaways¶

• Assume the network is unreliable and failures are partial; design every remote interaction for slowness, loss, duplication, and reordering.
• Decide consistency versus availability per operation using CAP/PACELC; document the model each flow provides.
• Idempotency, timeouts, bounded retries, and backoff-with-jitter are one package; never adopt retries without the other three.
• Circuit breakers and bulkheads contain failures; sagas with compensations replace unworkable distributed transactions.
• Treat delivery as at-least-once and make processing idempotent to achieve exactly-once effects.
• Correlated tracing, metrics, and logs are the only way to understand and operate distributed flows.

References and further reading¶

• Martin Kleppmann, Designing Data-Intensive Applications
• Andrew Tanenbaum and Maarten van Steen, Distributed Systems: Principles and Paradigms
• Michael Nygard, Release It!: Design and Deploy Production-Ready Software
• Sam Newman, Building Microservices
• Chris Richardson, Microservices Patterns (sagas, transactional messaging)
• Eric Brewer, "CAP Twelve Years Later" and Daniel Abadi on PACELC
• Leslie Lamport, "Time, Clocks, and the Ordering of Events in a Distributed System"
• Cindy Sridharan, Distributed Systems Observability
• Nassim Nicholas Taleb's notion of antifragility (as applied by resilience-engineering literature)