4.6 Distributed System Design Concepts

## Coordination & Configuration

In a distributed system, dozens or hundreds of services need to find each other, share configuration, and roll out changes safely.

Without coordination infrastructure, each service is an island that cannot discover its neighbors, adapt to changing conditions, or participate in system-wide decisions.

**Service Discovery and Registration**

When Service A needs to call Service B, it needs to know Service B's network address.

In a traditional deployment, you might hardcode an IP address or hostname in a configuration file.

In a distributed system where service instances are created and destroyed dynamically (by auto-scaling, deployments, and failure recovery), hardcoded addresses break constantly.

Service discovery maintains a registry of available service instances and their current addresses. It operates in two phases.

Registration: When a service instance starts, it registers itself with the discovery system: "I am an instance of the payment service, and my address is 10.0.3.42:8080." When the instance shuts down or fails a health check, it is deregistered.

Discovery: When another service needs to call the payment service, it queries the discovery system: "Give me the addresses of healthy payment service instances." The discovery system returns a list (10.0.3.42:8080, 10.0.3.87:8080, 10.0.4.12:8080), and the caller picks one (using client-side load balancing) or is routed to one (server-side discovery through a load balancer).

Kubernetes provides built-in service discovery through DNS.

A service named `payment-service` in the `default` namespace is reachable at `payment-service.default.svc.cluster.local`.

Kubernetes automatically updates the DNS entry as pods are created and destroyed.

For non-Kubernetes environments, dedicated discovery tools like Consul, etcd, and Eureka fill this role.

Health checking is tightly coupled with discovery.

A registered instance that is not actually healthy (it is up but returning errors) should be removed from the registry.

Active health checks (the discovery system pings instances periodically) and passive health checks (the discovery system monitors actual traffic success rates) both contribute to an accurate registry.

**Configuration Management at Scale**

Every service needs configuration: database connection strings, feature flag values, timeout settings, API keys for external services, and environment-specific parameters.

Managing configuration across hundreds of service instances in multiple environments (development, staging, production) is a coordination challenge.

Environment variables are the simplest approach. Each service reads its configuration from environment variables injected at startup. This works well for small systems but becomes unwieldy when you have hundreds of config values across dozens of services. Changing a value requires restarting the service.

Centralized configuration stores like etcd, Consul KV, or AWS Systems Manager Parameter Store hold configuration in a central location.

Services read their configuration from the store at startup and can watch for changes in real time.

When a value changes, services pick up the new value without restarting.

This enables dynamic configuration: changing a timeout from 5 seconds to 10 seconds across all instances of a service in seconds, without a deployment.

Configuration as code stores configuration in version-controlled files (YAML, JSON, or HCL) that are deployed alongside the application.

This provides auditability (every change is a commit with a message, author, and timestamp), review processes (configuration changes go through pull requests), and rollback capability (revert to a previous commit).

The trade-off is that changes require a deployment, which is slower than updating a centralized store.

Most production systems use a combination. Secrets go in a secrets manager (Chapter IV). Dynamic operational parameters (timeouts, rate limits, feature flags) go in a centralized config store.

Environment-specific settings (which database, which region) are baked into the deployment configuration.

**Feature Flags and Gradual Rollouts**

Feature flags (also called feature toggles) decouple deployment from release. You deploy code to production with the new feature behind a flag.

The feature is off by default. You turn it on for 1% of users, observe metrics, increase to 10%, observe again, and eventually roll it out to 100% or roll it back if something goes wrong.

Feature flags enable several patterns.

Canary releases expose a new feature to a small percentage of traffic before full rollout.

A/B testing shows different feature variants to different user groups and measures which performs better.

Kill switches instantly disable a feature in production without a deployment if it causes problems.

User targeting enables a feature for specific user segments (beta testers, internal employees, premium customers) before the general release.

Feature flag management tools include LaunchDarkly, Split.io, Unleash (open-source), and Flagsmith (open-source). These provide SDKs that evaluate flags in real time, dashboards for managing flags, and analytics for measuring the impact of each flag.

The risk with feature flags is technical debt.

Every flag adds a conditional branch to your code.

If flags are not cleaned up after they are fully rolled out, the codebase accumulates dead branches that make the code harder to read and maintain.

Establish a practice of removing flags within 30 days of full rollout.

**Distributed Coordination with ZooKeeper, etcd, Consul**

These three tools are the backbone of distributed coordination.

They have been covered in specific contexts throughout this handbook (leader election in Chapter III; service discovery earlier in this section), but understanding their broader role is valuable.

All three provide a strongly consistent, distributed key-value store that multiple services can use as a source of truth for shared state.

They enable leader election (ensuring one instance of a service is the designated leader), distributed locks (ensuring exclusive access to a shared resource), configuration storage (a central place for dynamic settings), service discovery (maintaining a registry of healthy service instances), and group membership (tracking which nodes are currently part of a cluster).

ZooKeeper is the most mature, with deep integration into the Hadoop and Kafka ecosystems. It uses the Zab consensus protocol and provides ephemeral nodes (which disappear when the creating client disconnects) and sequential nodes (for implementing locks and queues). ZooKeeper requires a dedicated cluster of typically 3 or 5 nodes.

etcd is simpler and more modern. It uses Raft consensus, provides a flat key-value space with a watch API for real-time change notifications, and is the coordination backbone of Kubernetes. etcd has a smaller operational footprint than ZooKeeper and is easier to set up.

Consul combines coordination with service discovery and health checking in a single tool. It uses Raft consensus, provides a key-value store, built-in DNS for service discovery, and multi-datacenter support. Consul is the best choice when you need both coordination and service discovery and want a single tool for both.

Interview-Style Question

> Q: Your system has 12 instances of a report generation service. A daily report must run exactly once. How do you coordinate this?

> A: Use leader election through etcd or Consul. All 12 instances compete to acquire a distributed lock (a key in etcd with a lease TTL). The instance that acquires the lock is the leader and runs the report. Other instances detect that the lock is held and skip the run. If the leader crashes mid-report, the lease expires, and another instance can acquire the lock on the next schedule. The report generation itself should be idempotent (safe to run again if it was partially completed before the crash). After the report completes, the leader releases the lock. The lock's TTL should exceed the maximum expected report duration so that the lock does not expire while the report is still running.

**KEY TAKEAWAYS**

* Service discovery maintains a live registry of available service instances. Health checks ensure only healthy instances receive traffic.

* Centralized configuration stores (etcd, Consul KV) enable dynamic config changes without redeployments. Combine with config-as-code for auditability. * Feature flags decouple deployment from release. They enable canary rollouts, A/B testing, and kill switches. Clean them up after full rollout to avoid technical debt. * ZooKeeper, etcd, and Consul provide strongly consistent coordination primitives: leader election, distributed locks, config storage, and service discovery. Choose based on your ecosystem.

## Consistency & Synchronization

When multiple nodes hold copies of the same data, they need mechanisms to detect conflicts, resolve disagreements, and converge toward a consistent state.

The techniques in this section are the building blocks that distributed databases, caches, and replication systems use to maintain order in a fundamentally unreliable environment.

**Vector Clocks and Version Vectors**

In a distributed system, you cannot rely on physical clocks to determine the order of events. Different machines have different clock drift.

Network delays mean events do not arrive in the order they occurred.

Two events might have identical timestamps but be completely unrelated.

Vector clocks solve this by tracking causal relationships between events rather than wall-clock time. Each node maintains a vector (an array) with one entry per node in the system.

When a node performs a local operation, it increments its own entry.

When a node sends a message, it includes its current vector. When a node receives a message, it merges the received vector with its own by taking the maximum value for each entry.

Consider a system with three nodes: A, B, and C. Each starts with vector \[0, 0, 0\].

Node A writes a value. Its vector becomes \[1, 0, 0\].

Node A sends the value to Node B.

Node B's vector becomes \[1, 1, 0\] (it merged A's \[1, 0, 0\] with its own \[0, 0, 0\] and incremented its own entry).

Node C independently writes a different value. Its vector becomes \[0, 0, 1\].

Now Node B and Node C have conflicting values.

By comparing their vectors, the system can determine that \[1, 1, 0\] and \[0, 0, 1\] are concurrent (neither dominates the other). This is a genuine conflict that requires resolution.

If instead Node C's vector were \[1, 1, 1\], it would dominate B's \[1, 1, 0\], meaning C's value is the newer one and there is no conflict.

Version vectors are a closely related concept used specifically for tracking the version history of a data item across replicas.

Amazon's Dynamo (the internal system that inspired DynamoDB) used version vectors to detect and resolve conflicts between replicas.

The practical takeaway: vector clocks and version vectors tell you whether two versions of data are causally related (one happened before the other) or concurrent (they happened independently and conflict).

This is far more reliable than comparing wall-clock timestamps, which can be wrong due to clock skew.

**CRDTs (Conflict-Free Replicated Data Types)**

CRDTs are data structures specifically designed to be replicated across multiple nodes and merged automatically without conflicts.

Every node can update its local copy independently, and when nodes sync, the CRDTs merge into a consistent state deterministically, with no coordination required.

The secret is that CRDTs are designed so that all possible merge operations are commutative (order does not matter), associative (grouping does not matter), and idempotent (applying the same update twice has no additional effect). These properties guarantee that regardless of the order in which updates and merges happen, all nodes converge to the same result.

Common CRDT types include the G-Counter (grow-only counter, each node increments its own counter, the total is the sum of all counters), PN-Counter (positive-negative counter, combines a G-Counter for increments and a G-Counter for decrements), G-Set (grow-only set, elements can be added but never removed), OR-Set (observed-remove set, elements can be added and removed, using unique tags to track each add operation), and LWW-Register (last-writer-wins register, concurrent updates are resolved by timestamp, with the latest timestamp winning).

CRDTs are used in collaborative editing (Figma uses CRDTs for real-time collaboration), distributed databases (Riak uses CRDTs for conflict resolution), and mobile/offline applications (CRDTs let users edit data offline, and changes merge correctly when connectivity returns).

The trade-off is that CRDTs are limited in the kinds of operations they can represent. Not every data model maps naturally to a CRDT.

And some CRDTs grow monotonically (like the OR-Set, which retains metadata about removed elements), consuming more memory over time.

Garbage collection of CRDT metadata requires coordination, which partially defeats their coordination-free appeal.

**Gossip Protocol**

Gossip protocols (also called epidemic protocols) spread information through a distributed system the way rumors spread through a social group.

Each node periodically shares what it knows with a randomly selected peer.

That peer incorporates the new information and shares it with another random peer.

Eventually, all nodes in the system learn the information, even though no central coordinator broadcast it.

Gossip is used for cluster membership (nodes discover which other nodes are alive and which have failed), data dissemination (spreading configuration updates or metadata across all nodes), failure detection (if a node stops gossiping, its peers suspect it has failed and share that suspicion until the cluster agrees), and aggregate computation (nodes can gossip their local metrics and each node computes a system-wide aggregate).

Cassandra uses gossip for cluster membership and failure detection. Each node gossips with a few random peers every second, sharing its knowledge of the cluster state.

Within seconds, a node failure is detected and propagated across the entire cluster without any central failure detector.

Gossip has several appealing properties. It is decentralized (no single point of failure), scalable (each node communicates with a fixed number of peers regardless of cluster size), and fault-tolerant (information propagates even when some nodes are unreachable).

The trade-off is eventual propagation: information does not reach all nodes instantly. There is a short window (seconds) where different nodes have different information.

**Quorum-Based Reads and Writes**

Quorum-based systems require a minimum number of nodes to agree before a read or write is considered successful. This provides tunable consistency without requiring all nodes to participate in every operation.

For a system with N replicas, you configure a write quorum W (the number of replicas that must acknowledge a write) and a read quorum R (the number of replicas that must respond to a read).

The fundamental rule is: if W \+ R \> N, every read is guaranteed to see the latest write, because the read quorum and write quorum overlap by at least one node.

With N=3, W=2, R=2: a write succeeds when 2 of 3 replicas acknowledge.

A read succeeds when 2 of 3 replicas respond. Since 2 \+ 2 \= 4 \> 3, there is always at least one replica in the read set that participated in the latest write. That replica returns the latest value.

Tuning W and R trades consistency for availability and latency. Setting W=1, R=3 gives fast writes (only one replica needed) but slow reads (all three must respond). Setting W=3, R=1 gives slow writes but fast reads. Setting W=2, R=2 balances both.

Cassandra and DynamoDB both support tunable quorum configurations, letting you choose consistency guarantees per operation. A payment write might use W=majority for strong consistency, while a product catalog read might use R=1 for low latency with eventual consistency.

Interview-Style Question

> Q: Your distributed database has 5 replicas. You want reads to always return the latest write while tolerating the failure of up to 2 replicas. What quorum configuration should you use?

> A: Use W=3 and R=3 (majority quorums). With W+R \= 6 \> 5, every read is guaranteed to overlap with the write quorum, ensuring strong consistency. With W=3, a write succeeds as long as 3 of 5 replicas acknowledge, tolerating 2 failures. With R=3, a read succeeds as long as 3 of 5 replicas respond, also tolerating 2 failures. Both operations can survive any 2 replica failures while maintaining strong consistency. If you need to tolerate 2 failures with lower latency reads, you could use W=3, R=2 (W+R=5=N), but this only guarantees overlap when all nodes are healthy. With one node down, R=2 might miss the latest write. Sticking with R=3 provides the strongest guarantee under your failure tolerance requirements.

_Quorum Read/Write_

### KEY TAKEAWAYS

* Vector clocks track causal relationships between events, detecting whether two data versions are causally ordered or concurrent (conflicting). They are more reliable than wall-clock timestamps.

* CRDTs are data structures that merge automatically without conflicts. They enable coordination-free replication for counters, sets, and registers, powering collaborative editing and offline-capable applications. * Gossip protocols spread information decentrally through random peer-to-peer communication. They handle cluster membership, failure detection, and data dissemination without a central coordinator. * Quorum-based systems require W writes and R reads to overlap (W \+ R \> N) for strong consistency. Tuning W and R trades consistency for latency and availability.

## Distributed Patterns

Distributed systems face recurring challenges: how to add capabilities without modifying existing services, how to migrate from old systems to new ones without downtime, and how to maintain data consistency across service boundaries.

The patterns in this section provide proven solutions to these challenges.

**Sidecar Pattern**

The sidecar pattern attaches an auxiliary process to a service instance.

The sidecar runs alongside the main application in the same host or pod, handling cross-cutting concerns that the application should not implement itself.

You encountered the sidecar pattern in Chapter II, (service mesh). Envoy runs as a sidecar proxy next to each microservice, handling mutual TLS, service discovery, load balancing, retries, and observability. The application knows nothing about these concerns. It makes plain HTTP calls to localhost, and the sidecar handles everything else transparently.

Beyond service mesh proxies, sidecars are used for log collection (a Fluentd sidecar collects and forwards application logs without the application knowing about the logging infrastructure), configuration refresh (a sidecar watches a configuration store and writes updated config files that the application reads), certificate management (a sidecar handles TLS certificate rotation and presents fresh certificates to the application), and monitoring agents (a Datadog or Prometheus agent runs as a sidecar, scraping the application's metrics endpoint).

The sidecar pattern works because it keeps the application focused on business logic while delegating infrastructure concerns to a dedicated, reusable component.

The trade-off is resource overhead: each sidecar consumes CPU and memory, and across hundreds of service instances, this overhead adds up.

**Ambassador Pattern**

The ambassador pattern places a proxy between a service and the external resources it depends on.

The proxy handles the complexity of connecting to those resources: retries, circuit breaking, connection pooling, routing, and protocol translation.

The difference from the sidecar pattern is focus.

A sidecar handles inbound concerns (how traffic reaches the service).

An ambassador handles outbound concerns (how the service reaches external dependencies).

For example, a service that connects to a Redis cluster needs to handle connection pooling, failover between Redis nodes, and retry logic.

Instead of building this into every service that uses Redis, you deploy an ambassador proxy that handles the Redis connection complexity.

The application connects to `localhost:6379` (the ambassador), and the ambassador manages the actual connection to the Redis cluster.

Ambassadors are also used for connecting to external APIs (handling rate limiting, retries, and authentication), connecting to databases across regions (routing writes to the primary and reads to the nearest replica), and translating protocols (the application speaks HTTP, the ambassador translates to gRPC for the downstream service).

**Strangler Fig Pattern (for Migrations)**

The strangler fig pattern was introduced in Chapter III, in the context of monolith-to-microservices migration. It applies more broadly to any incremental system replacement.

The pattern works by placing a routing layer (API gateway, reverse proxy) in front of both the old system and the new system. Initially, all traffic goes to the old system.

As you rebuild individual components in the new system, you update the routing layer to direct traffic for those components to the new implementation.

Over time, more and more traffic flows to the new system.

Eventually, the old system handles no traffic and can be decommissioned.

The strangler fig pattern applies to database migrations (route queries to the old database or the new one based on the table or operation), API version transitions (route v1 API calls to the legacy service and v2 calls to the new service), and infrastructure migrations (route traffic to the old data center or the new cloud region based on readiness).

The critical property is that the migration is incremental and reversible.

If a new component has problems, you route its traffic back to the old system. No big-bang cutover. No all-or-nothing risk.

**Outbox Pattern**

The outbox pattern solves a specific but common problem in microservices: how do you reliably update a database and publish an event at the same time?

Consider an order service that inserts an order record into its database and publishes an "OrderPlaced" event to Kafka.

If the database write succeeds but the Kafka publish fails (network issue, Kafka is briefly unavailable), the order exists in the database but the event is lost.

Downstream services never learn about the order.

You might think: just wrap both operations in a transaction. But a database transaction cannot span the database and Kafka. They are independent systems with independent transaction mechanisms.

The outbox pattern solves this by writing the event to an "outbox" table in the same database as the order record, within the same transaction. Both writes succeed or both fail atomically (they are in the same database).

A separate process (the outbox publisher) reads new entries from the outbox table and publishes them to Kafka.

Once published, the entry is marked as sent.

The outbox publisher can be a polling process that periodically queries the outbox table for unsent entries, or a Change Data Capture (CDC) connector (like Debezium) that captures the outbox inserts from the database's transaction log and publishes them to Kafka in near-real-time.

The outbox pattern guarantees that if the database record exists, the event will eventually be published. The event might be published more than once (if the publisher crashes after publishing but before marking the entry as sent), which is why consumers must be idempotent (a recurring theme throughout this handbook).

_The Outbox Pattern_

### Anti-Corruption Layer

When your system integrates with an external system (a third-party API, a legacy internal system, or an acquired company's platform), the external system's data model and API contracts rarely match yours.

If you adapt your internal model to match the external system, the external system's design decisions, naming conventions, and quirks leak into your codebase.

Over time, your system becomes coupled to the external system's structure, making it painful to change either one.

The anti-corruption layer (ACL) is a translation boundary between your system and the external system. It converts the external system's data model into your internal model and vice versa. Your services interact only with the ACL, which speaks your language.

The ACL handles the translation to and from the external system's language.

For example, your order system represents amounts in cents as integers (12500 for $125.00). A legacy billing system represents amounts as strings with currency symbols ("$125.00").

The ACL translates between these representations, so your order system never deals with the string format and the billing system never sees your integer format.

ACLs are especially valuable during migrations (the ACL sits between the new system and the legacy system, translating between their models), third-party integrations (the ACL isolates your system from changes to the external API; when the third party changes their API, you update the ACL, not your entire system), and mergers and acquisitions (each company's system speaks its own language, and the ACL translates between them until the systems are unified).

The ACL can be implemented as a service (a dedicated translation microservice), as a library (a translation layer within the calling service), or as an API gateway rule (transforming requests and responses at the routing layer).

**Beginner Mistake to Avoid**

New engineers sometimes skip the outbox pattern and publish events directly after the database write, outside the transaction.

"The database write succeeded, so the event should publish fine."

This works 99.9% of the time. The 0.1% when Kafka is momentarily unreachable or the service crashes between the database commit and the Kafka publish creates silent data inconsistencies that are extremely hard to debug.

The outbox pattern adds a small amount of complexity but eliminates an entire class of distributed consistency bugs.

If your system uses events for inter-service communication and data consistency matters, use the outbox pattern.

Interview-Style Question

> Q: You are replacing a legacy monolithic order management system with new microservices. The legacy system handles 10,000 orders per day. You cannot afford downtime during the migration. How do you approach this?

> A: Use the strangler fig pattern. Place an API gateway or reverse proxy in front of both the legacy system and the new microservices. Initially, all order traffic routes to the legacy system. Build the new Order Service and test it thoroughly with production-like traffic in a staging environment. When ready, route a small percentage of traffic (5%) to the new service and the rest to the legacy system. Monitor error rates, latency, and data correctness. Gradually increase the percentage as confidence grows: 5% → 25% → 50% → 100%. At each step, you can instantly route back to 0% if issues arise. For the data layer, use the anti-corruption layer pattern: the new Order Service speaks its own clean data model, and an ACL translates between the new model and the legacy system's model during the transition period when both systems operate simultaneously. Use the outbox pattern in the new Order Service to reliably publish order events for downstream services. Once 100% of traffic flows to the new service with no issues for a stability period (typically 2 to 4 weeks), decommission the legacy system.

**KEY TAKEAWAYS**

* The sidecar pattern attaches an auxiliary process for cross-cutting concerns (logging, TLS, monitoring) alongside each service instance. It keeps application code focused on business logic.

* The ambassador pattern places an outbound proxy between a service and its external dependencies, handling retries, connection pooling, and protocol translation. * The strangler fig pattern enables incremental, reversible migration from old systems to new ones by routing traffic progressively through a proxy layer. * The outbox pattern guarantees that database writes and event publications stay in sync by writing events to a database table within the same transaction and publishing them separately. * The anti-corruption layer translates between your system's data model and an external system's model, preventing external design decisions from leaking into your codebase. * These patterns are building blocks that combine in real systems. A migration might use the strangler fig pattern for traffic routing, the anti-corruption layer for data model translation, and the outbox pattern for reliable event publishing.