Capstone: designing a geo-distributed service

This final lesson assembles the course into a design. It begins with the large-scale storage and processing systems whose papers shaped the field, from the Google File System to Spanner, then looks at the control-loop architecture of cloud-native platforms such as Kubernetes, and at the legal side of placing data in more than one country. It ends with a complete design for Ledgerline, the payments ledger followed throughout the course, deployed across several data centers, with every choice tied back to the lesson that justifies it.

Distributed storage systems

The Google File System (Ghemawat, Gobioff and Leung, 2003) showed how to build a reliable file system from thousands of unreliable commodity machines. Files were split into 64 MB chunks, each replicated on three chunkservers, with a single master holding metadata in memory and an operation log replicated for recovery. GFS relaxed consistency, allowing concurrent appends to produce duplicated or padded records that applications had to tolerate, in exchange for simplicity and throughput. The Hadoop Distributed File System (HDFS) followed the same design in open source. Bigtable (2006) layered a sorted, range-partitioned map of rows, columns and timestamps on GFS, with tablets served by tablet servers and Chubby for master election; HBase and, in part, Cassandra adopted its data model. Amazon's Dynamo (2007) took the opposite approach, a leaderless, always-writable key-value store using consistent hashing, sloppy quorums, vector clocks and Merkle-tree anti-entropy (lessons 3, 6 and 10), built for a shopping cart that must accept writes even during failures. Cassandra (2008) combined Dynamo's distribution with Bigtable's data model. Spanner (2012) combined range partitioning, Paxos-replicated groups, two-phase commit and TrueTime (lessons 3, 5, 8) into a globally distributed database with strictly serializable transactions.

Distributed data processing

Storing data across thousands of machines raised the question of how to compute over it. MapReduce (Dean and Ghemawat, 2004) gave a simple answer. A programmer writes a map function that turns each input record into key-value pairs and a reduce function that combines all values for one key. The framework splits the input, runs map tasks near the data, partitions their output by key, shuffles it to reduce tasks and writes the results. Fault tolerance is the elegant part: map and reduce functions are deterministic and have no side effects, so a failed task is simply re-executed on another machine. Because map output is stored on the mapper's local disk, a crashed worker's completed map tasks must also be re-run. Slow machines, stragglers, are handled by launching backup copies of the last few tasks and taking whichever finishes first, the same idea as hedged requests in lesson 1.

# MapReduce word count on a simulated cluster, with a worker failing mid-job.
from collections import defaultdict
DOCS = [
    "payment recorded payment refunded", "invoice paid invoice overdue",
    "payment recorded invoice paid", "refund issued payment reversed",
    "invoice overdue reminder sent", "payment recorded receipt sent",
]
R = 3                                                    # number of reduce partitions
WORKERS = ["w1", "w2", "w3"]

def map_task(doc):                                       # emits (word, 1) pairs, partitioned by hash
    parts = defaultdict(list)
    for word in doc.split():
        parts[sum(map(ord, word)) % R].append((word, 1))  # deterministic hash for the demo
    return parts

def reduce_task(pairs):
    counts = defaultdict(int)
    for word, n in pairs: counts[word] += n
    return counts

# Map phase: the master assigns map tasks round robin
completed, executions = {}, 0
for task, doc in enumerate(DOCS):
    worker = WORKERS[task % len(WORKERS)]
    completed[task] = (worker, map_task(doc)); executions += 1
print("map tasks done:", {t: w for t, (w, _) in completed.items()})

# Before the reduce phase starts, w2 crashes. Its map output was on its local disk,
# so every map task it completed is lost and must be re-executed elsewhere.
WORKERS.remove("w2")
lost = [t for t, (w, _) in completed.items() if w == "w2"]
for i, t in enumerate(lost):
    worker = WORKERS[i % len(WORKERS)]
    completed[t] = (worker, map_task(DOCS[t])); executions += 1
print(f"w2 crashed; re-executed map tasks {lost} on {WORKERS}")

intermediate = defaultdict(list)
for worker, out in completed.values():
    for p, pairs in out.items(): intermediate[p].extend(pairs)

# Reduce phase
total = {}
for p in range(R):
    total.update(reduce_task(intermediate[p]))
print(f"map task executions: {executions} for {len(DOCS)} tasks")
print("counts:", dict(sorted(total.items(), key=lambda kv: (-kv[1], kv[0]))))

map tasks done: {0: 'w1', 1: 'w2', 2: 'w3', 3: 'w1', 4: 'w2', 5: 'w3'}
w2 crashed; re-executed map tasks [1, 4] on ['w1', 'w3']
map task executions: 8 for 6 tasks
counts: {'payment': 5, 'invoice': 4, 'recorded': 3, 'overdue': 2, 'paid': 2, 'sent': 2, 'issued': 1, 'receipt': 1, 'refund': 1, 'refunded': 1, 'reminder': 1, 'reversed': 1}

Apache Spark (Zaharia and colleagues, 2010) generalized MapReduce into a dataflow of transformations on resilient distributed datasets, keeping intermediate data in memory and recovering lost partitions by recomputing them from their lineage, the recorded sequence of transformations, rather than by replication. Stream processors such as Apache Flink, Kafka Streams and Google's Dataflow (based on the 2015 Dataflow Model paper) process unbounded event streams continuously, using event-time windows and watermarks to handle late, out-of-order data.

Exactly-once processing, precisely

Stream processors advertise exactly-once processing, which seems to contradict lesson 2's claim that exactly-once delivery is impossible. The resolution is that they guarantee exactly-once effects on their own state, not exactly-once execution. Flink takes consistent distributed snapshots of operator state, using barriers in the streams like Chandy–Lamport markers (lesson 4), together with the source offsets; after a failure it restores the snapshot and replays input from those offsets. Records are processed more than once, but the restored state reflects each exactly once. Kafka's transactions do the same for read-process-write pipelines within Kafka by committing outputs and consumer offsets atomically. Once results leave the system, for example to send an email or call a payment network, exactly-once requires the external system to be idempotent or transactional, which brings back idempotency keys and two-phase commit. The phrase is best read as "effectively once, within these boundaries".

Cloud-native control loops

Kubernetes, released by Google in 2014 and now the standard platform for running containers, is itself a distributed system built on ideas from this course. Its entire state, the desired configuration of every deployment, service and pod, lives in etcd (lesson 9). Around it run controllers, each an independent loop that repeatedly observes the actual state of the cluster, compares it with the desired state, and acts to reduce the difference: starting pods that are missing, stopping extra ones, replacing pods on failed nodes. This reconciliation pattern is level-triggered rather than edge-triggered: a controller does not need to see every event, because each pass recomputes what to do from current state, so missed or duplicated notifications, restarts of the controller and partial failures are all tolerated. Leader election among controller replicas uses leases in etcd with fencing by revision (lesson 4).

# A Kubernetes-style control loop: observe actual state, compare with desired state, act.
import random
random.seed(450)
desired = {"ledger-api": 4, "invoice-api": 3}
actual = {"ledger-api": ["p1", "p2", "p3", "p4"], "invoice-api": ["q1", "q2", "q3"]}
next_id = 5

def reconcile():
    global next_id
    actions = []
    for svc, want in desired.items():
        have = actual[svc]
        while len(have) < want:                          # too few: start replicas
            name = f"{svc[0]}{next_id}"; next_id += 1
            have.append(name); actions.append(f"start {svc}/{name}")
        while len(have) > want:                          # too many: stop replicas
            actions.append(f"stop {svc}/{have.pop()}")
    return actions or ["no change"]

for step in range(5):
    if step == 1:
        dead = random.sample(actual["ledger-api"], 2)    # a node fails, taking two pods with it
        for d in dead: actual["ledger-api"].remove(d)
        print(f"step {step}: node failure removed {dead}")
    if step == 3:
        desired["invoice-api"] = 5                       # an operator scales up
        print(f"step {step}: desired invoice-api replicas changed to 5")
    print(f"step {step}: reconcile ->", "; ".join(reconcile()))

step 0: reconcile -> no change
step 1: node failure removed ['p1', 'p2']
step 1: reconcile -> start ledger-api/l5; start ledger-api/l6
step 2: reconcile -> no change
step 3: desired invoice-api replicas changed to 5
step 3: reconcile -> start invoice-api/i7; start invoice-api/i8
step 4: reconcile -> no change

Around the services themselves, a service mesh such as Istio or Linkerd places a proxy beside each service instance to handle the communication concerns of lessons 2 and 11 uniformly: mutual TLS between services, retries with budgets, timeouts, circuit breaking, load balancing and telemetry. Moving these into infrastructure makes them consistent across services written in different languages, at the cost of extra latency per hop and another complex system to operate.

Geo-distribution

Running in several regions improves latency for distant users and survives the loss of a whole region, but it brings the physics of lesson 1 to the fore. A round trip between Montréal and Calgary, or between Ohio and Oregon, takes tens of milliseconds, so any operation that needs a majority across regions pays that cost. Designers choose among several patterns. In active-passive, one region serves traffic and replicates asynchronously to a standby; failover loses recent writes and takes minutes. In active-active with partitioned ownership, each region is the leader for some data, such as the businesses located near it, and serves it with local latency, while other regions hold replicas. In global consensus, each data group's replicas span regions so that a regional outage loses no committed write, and commits pay a cross-region round trip, as in Spanner and CockroachDB's multi-region configurations. And for data that can merge, multi-leader replication with CRDTs (lesson 7) gives local writes everywhere.

# Latency budget for Ledgerline's US deployment (round trips are illustrative).
RTT = {("OHI", "VIR"): 12, ("OHI", "ORE"): 50, ("VIR", "ORE"): 62}
def rtt(a, b): return 1 if a == b else RTT.get((a, b)) or RTT[(b, a)]   # 1 ms between zones of one region

def commit_latency(leader, replicas):
    # Raft/Paxos commit: one round trip from the leader to the fastest majority (leader counts)
    others = sorted(rtt(leader, r) for r in replicas[1:])  # replicas[0] is the leader itself
    need = len(replicas) // 2                            # acknowledgements needed besides the leader
    return others[need - 1]

for name, reps in [("3 replicas in Ohio zones", ["OHI"] * 3),
                   ("Ohio, Virginia, Oregon", ["OHI", "VIR", "ORE"]),
                   ("2 Ohio + 2 Virginia + Oregon", ["OHI", "OHI", "VIR", "VIR", "ORE"])]:
    degraded = list(reps)
    if "VIR" in degraded: degraded.remove("VIR")      # one VIR replica fails
    region_ok = max(reps.count(r) for r in set(reps)) <= len(reps) // 2
    print(f"{name:32s} commit {commit_latency('OHI', reps):2d} ms; "
          f"with one VIR replica down {commit_latency('OHI', degraded):2d} ms; "
          f"survives losing a region: {'yes' if region_ok else 'no'}")

3 replicas in Ohio zones         commit  1 ms; with one VIR replica down  1 ms; survives losing a region: no
Ohio, Virginia, Oregon           commit 12 ms; with one VIR replica down 50 ms; survives losing a region: yes
2 Ohio + 2 Virginia + Oregon     commit 12 ms; with one VIR replica down 12 ms; survives losing a region: yes

The placements show the central trade-off. Keeping all replicas in one region's availability zones gives commits of about a millisecond but loses the whole group if the region fails. Spreading three replicas across three regions survives a regional failure but pays a cross-region round trip on every commit, and if the replica in the nearer region fails, commits wait for the distant third region. The five-replica layout, with two replicas in each of two nearby regions and one far away, commits at the nearby round trip, keeps that latency when any single replica fails, and still survives the loss of any one region. That resilience to ordinary machine failures, at the cost of two extra replicas, is why five-replica groups spread this way are a common choice for critical data.

Data residency and sovereignty

Where data may be stored is a legal as well as technical question. The United States has no general data-localization law for private companies, but several rules affect placement. The CLOUD Act of 2018 lets US authorities compel a provider subject to US jurisdiction to produce data in its possession, custody or control wherever it is stored, which foreign customers consider when choosing providers. Federal agencies buy cloud services authorized under FedRAMP, and sector rules such as HIPAA for health information set security requirements, though not strict residency. Since April 2025 a Department of Justice rule restricts transfers of bulk sensitive personal data to designated countries of concern. Customers in other countries bring their own rules: Canadian clients, for example, may be subject to Quebec's Law 25, which requires a privacy impact assessment before personal information leaves Quebec, and the EU's GDPR restricts transfers out of the European Union. For Ledgerline these rules shape placement: each customer's ledger is partitioned by business and its replicas pinned to regions that satisfy that customer's obligations, with US customers' data kept in US regions by default.

Designing Ledgerline

We can now put the pieces together. Ledgerline's requirements are: never lose an acknowledged payment; never record a payment twice; keep each business's balance correct, with no overdrafts beyond its limit; stay available through the loss of a machine, a zone or a whole region; give users in each country low latency; and keep each business's data in an appropriate jurisdiction. The design below satisfies them, and each choice comes from an earlier lesson.

A payment then flows as follows. The client generates an idempotency key and calls the regional API service with a deadline. The API service looks up the business's partition in a cached copy of the partition map, kept authoritative in etcd, and forwards the request to that partition's Raft leader. The leader checks the idempotency key, verifies the balance rule, and appends a log entry containing the payment, the key and an outbox event; the entry commits when a majority of the five replicas, in at least two regions, have it durably. The API service replies. Asynchronously, a relay publishes the outbox event to a Kafka topic partitioned by business, and consumers send receipts, update reporting replicas and score fraud, each deduplicating by event identifier. If a region fails, partitions whose leaders were there elect new leaders among the surviving replicas within seconds, without losing committed entries; clients retry with the same idempotency keys; and the error budget absorbs the brief unavailability.

Every design involves trade-offs, and this one is no exception. Payments pay a cross-region round trip, tens of milliseconds, for regional fault tolerance, a PACELC choice of consistency over latency. Draft invoices and view counters, which do not need it, use cheaper, local, AP storage. A business with an enormous transaction rate could outgrow a single partition and need its hot accounts split, at the cost of cross-partition transactions. And the system's correctness rests on components whose subtle bugs are hard to find, so it is tested with deterministic simulation and fault injection before it is trusted with money.

Where to go next

Distributed systems is a field where practice and research meet closely. The original papers cited throughout this course, Lamport on time, Chandy and Lamport on snapshots, FLP, Paxos Made Simple, Raft, Dynamo, Bigtable, Spanner, MapReduce and The Tail at Scale, are readable and rewarding. Martin Kleppmann's Designing Data-Intensive Applications and van Steen and Tanenbaum's Distributed Systems are excellent textbooks. Published postmortems from cloud providers and the Jepsen analyses show how real systems fail. And for building, AI-480 applies these ideas to distributed training, AI-420 to machine learning systems in production, ITIL-141 to service management, and the DCE program to the physical data centers underneath it all.

Worked examples