Distributed Systems ยท Chapter 03

One transaction, many machines

Once your data lives on many shards, a single purchase can touch several of them at once โ€” and there's no shared lock to keep it atomic. This chapter covers the two ways to pull that off, why real companies avoid it when they can, and how they scale it when they can't.

๐Ÿงญ
How to read this

The chapter's spine is one repeated lesson: avoid cross-shard transactions when you can; when you can't, coordinate carefully. It opens with the two coordination mechanisms (2PC and Saga) and how to choose, then shows how sharding forces you to rewrite application code, then grounds it in how Dropbox, Slack, and Square actually scale to millions of QPS โ€” and closes with the off-the-shelf answers (distributed SQL, ACID-capable NoSQL). Anything marked [+] is context I added; everything else is rewritten from the ten teaching lessons.

01

Two-Phase Commit: ask everyone, then commit

When one atomic transaction must span several nodes, the classic protocol is Two-Phase Commit (2PC) โ€” it runs the transaction in a prepare phase and a commit phase, coordinated by one node.

  1. Prepare phase. A designated coordinator asks every participant node whether it can carry out the transaction โ€” can it acquire the needed locks, does it have capacity? Each node does everything needed for success except the final commit, and replies ready.
  2. Commit phase. Once all nodes confirm readiness, the coordinator tells them to commit. Each commits and sends an acknowledgment; when all acks arrive, the coordinator completes the transaction. Any failure in any phase โ†’ the whole thing rolls back and locks release.
Coordinator Participant + write-ahead log Participant + WAL Participant + WAL 1ยท prepare? 2ยท ready โ†’ commit โ†’ ack
Figure 1 โ€” Two-Phase Commit: the coordinator polls all participants (prepare), then orders the commit once everyone is ready. Each node keeps a write-ahead log. Original diagram after Figure 2.2.

Write-ahead log (WAL)

Every participant records its activity to a write-ahead log in stable storage before committing. If a node crashes mid-transaction, on recovery it consults the log to learn whether its last operation succeeded, partially ran, or never ran. Logs are append-only (all do/undo/redo entries in chronological order), so no record is lost across failures; checkpoints mark where updates are permanently written and the log is cleared.

โ›“๏ธ
2PC's cost: it blocks

2PC guarantees strong consistency (it uses locks) but is blocking. If a participant goes down mid-transaction, everyone โ€” including the coordinator โ€” waits, holding locks, until it returns. Transaction latency is bound by the slowest node, and if the coordinator dies it's a single point of failure. The protocol assumes each node keeps a WAL and no node stays down forever. When blocking hurts too much, teams reach for Saga.

02

The Saga pattern: a chain of undoable steps

Saga runs a distributed transaction as a sequence of local transactions, each with a compensating (undo) transaction that's idempotent. Either all steps succeed in order, or on failure the compensating transactions roll back from the point it broke. The bookstore purchase flow:

  1. Create order โ€” compensate: cancel order.
  2. Process payment (from card info) โ€” compensate: refund.
  3. Update inventory โ€” compensate: restock.
  4. Kick off delivery.

Each step is a separate transaction with its own compensating counterpart. A coordinator logs every step; on failure it triggers the compensating transactions โ€” and because a compensating call might itself fail and be retried, those transactions must be idempotent. Unlike 2PC (a database protocol plus custom code), Saga is a pattern the developers design and implement โ€” you own the rollbacks and compensations. Frameworks like Apache Camel help.

Two ways to implement Saga

Approach How it coordinates
Choreography No central coordinator โ€” event-based. Each service executes its transaction, publishes an event, and services subscribe to each other's events to decide what happens next. Fine for a few events; becomes a tangled mess with many microservices.
Orchestration A dedicated orchestrator (embedded in a service or standalone) manages the whole chain, calling each microservice via its API. Better separation of concerns โ€” services don't track each other's events.
Choreography order payment delivery inventory events, no controller Orchestration orchestrator order payment inventory delivery
Figure 2 โ€” Saga implementations: choreography (services react to each other's events) vs. orchestration (one orchestrator drives every service). Original diagram after Figures 2.4โ€“2.5.
03

2PC or Saga? There's no formula

Two-Phase Commit Saga
Nature Blocking; uses DB locks. Non-blocking; linear sequence, no locks.
Consistency Strong (by default). Leans eventual; strong is possible but tricky & bug-prone.
DB support Every DB involved must support the 2PC protocol. Works even with DBs that don't support transactions.
Duration Must release locks fast โ€” short transactions. No locks, so a transaction can run seconds to days.
Shape Whole transaction commits in one request. Big transaction split into small sequential local ones.
Fits Strong-consistency needs within controllable nodes. Multi-step flows: flight/hotel booking, e-commerce.
๐Ÿฆ
The question that settles it

For a $100 wire transfer from Bank A to Bank B โ€” would Bank A really hold a lock on Bank B's database to guarantee strong consistency via 2PC? Obviously not. Saga fits better here, even though strong consistency is harder with it. The point: there's no definite formula โ€” the choice depends on the business use case, application design, tech stack, and more.

04

Sharding forces you to rewrite the code

Once you shard, the application can no longer pretend there's one database โ€” the code must know where data lives and route each query to the right shard. As nodes are added/removed and data is resharded and moved, that routing logic must keep up. Sharding often means real schema changes and refactoring, and demands you design around your frequent queries.

๐Ÿšซ
The golden rule: avoid cross-shard transactions

Keep them minimal or eliminate them. Watch the corner cases โ€” they're where production breaks and you end up compensating customers. And doing ACID in your application code instead of relying on the database's battle-tested implementation takes serious skill and resources. Avoid cross-shard transactions unless absolutely necessary.

The banking cross-shard query problem

A bank's MoneyTransfer table has ~45M entries/day across 200 shards. Query: "fetch โ‰ฅ9000 transactions with transfer amount below $50 for a given day." Each shard is independent and knows nothing of the others. Two patterns handle this:

API Composition An API composer fires the query at all shards, pulls results into memory, and joins/sorts there. Fine when results fit in RAM โ€” otherwise you're into disk-based sorting and custom logic. Same idea as fetching from multiple microservices and joining in-app.
CQRS (Command Query Responsibility Segregation) A dedicated read-only query shard holds redundant, pre-computed results kept in sync via an async write process. The heavy query hits one shard with answers ready โ€” huge latency win, at the cost of possible staleness and sync complexity. Often paired with Event Sourcing.
05

Real world: Dropbox at 10M requests/second

Dropbox stores petabytes of metadata behind a system called Edgestore โ€” an abstraction over thousands of MySQL instances across data centers, holding trillions of entries, serving millions of QPS with strongly consistent low-latency reads/writes. Edgestore hides the physical sharding from the app.

Their tweaks to 2PC

They also co-locate frequently accessed data so only 5โ€“10% of transactions span multiple shards. Edgestore itself acts as a shard manager, owning the operational overhead โ€” resharding under load, managing slow queries, and more.

๐Ÿงฐ
The universal pattern here

Almost every large system running relational clusters uses a shard manager โ€” built in-house (Dropbox's Edgestore) or off-the-shelf (Vitess, ProxySQL, Citus). It absorbs shard-management operations and makes the app feel like it talks to a single database โ€” which is exactly what the next lessons show at Slack and Square.

06

Slack, Square & the Vitess shard manager

Vitess is the open-source shard manager (built by YouTube, run by Square, Slack, Pinterest, Flipkart, Airbnb, GitHub, Roblox, Twitter, and more). It sits between the app and MySQL shards, adding built-in sharding with minimal app-side logic, plus connection pooling, automatic failover/backup, query rewriting, and live resharding.

Slack

MySQL is Slack's primary store; Vitess serves 99% of query load โ€” 2.3M QPS peak (2M reads, 300K writes, ~2ms latency). They started on a LAMP stack with three clusters โ€” Shards (customer data partitioned by workspace id, so all of a workspace's data lives together), a Metadata cluster (maps workspace id โ†’ shard), and a Kitchen sink cluster (non-workspace data). The problem: some workspaces outgrew a single shard, creating hotspots โ€” a few big customers overloaded their shards while other hardware sat idle, and a single shard failing meant an outage for that customer. The fix: move all sharding logic out of the monolith into Vitess (now split by keyspace, not workspace id), making the logic reusable across services.

app vtgate vtgate stateless proxies shard +vttablet shard +vttablet shard +vttablet ZooKeeper / etcd topology coordination
Figure 3 โ€” Vitess architecture: stateless vtgate proxies route queries to MySQL shards (each with a vttablet), coordinated via ZooKeeper/etcd. Original diagram after Figures 3.1โ€“3.2.

Square (Cash App)

The Cash App grew from a hack-week single MySQL instance to half a million lines of code, forcing a shard โ€” also on Vitess. Because it processes money, ACID and strong consistency were vital, so they had to avoid cross-shard money transactions. Their solution: entity groups โ€” the customer is the main entity group, and Vitess is configured so all members of an entity group stay on the same shard. Transactions inside one entity group land on one shard and are ACID-compliant; those spanning entity groups aren't. To enable this, they remodeled from a normalized schema to a partitionable one.

07

Finding the shard when the query lacks the key

Square's entity-group model has a root entity (customer) and child entities that carry the customer id to stay associated โ€” e.g. a customer row (root) and its credit-card rows (children, each with a CustomerId column). A query with the customer id routes cleanly:

SELECT * FROM Customer WHERE CustomerId = 752
                

But what about a query that only has a credit-card id?

SELECT * FROM CreditCard WHERE CardId = 24
                

Vitess can't know which shard holds it, and scanning all shards is unacceptable. Adding CustomerId to every query works for some but breaks when you have only card ids and need to find their owners โ€” and ~25% of Cash App's DB traffic had no customer id, too much to fix by hand.

๐Ÿ—‚๏ธ
VIndex โ€” a distributed index

Vitess's VIndex is a distributed index backed by a dedicated table (e.g. CreditCardLookUp holding both CreditCardId and CustomerId ). Just as a table index locates the right row, a distributed index locates the right shard โ€” so Vitess resolves card-only queries to their owning customers without scanning every shard. The catch: whenever CreditCard changes, the lookup table must be kept in sync, and developers must guard against inconsistency.

The through-line across Dropbox, Slack, and Square: understanding your application's queries is the prerequisite to sharding well.

08

Distributed SQL: buy the whole thing pre-built

Distributed SQL is a single relational database whose data is replicated across nodes worldwide, using consensus algorithms (Raft, Paxos) to agree on values and offering an SQL-compatible distributed query layer. Examples: Google Spanner, CockroachDB, YugabyteDB, TiDB, Amazon Aurora. Architecturally similar to a Vitess setup โ€” the key difference is these are built from the ground up to scale rather than layered onto an existing engine (Vitess for MySQL, Citus for Postgres).

Distributed SQL offers Note
ACID transactions At least within the availability zone where the write originates.
Global replication With strong consistency.
Relational structure Tables, rows, columns.
Automatic sharding + shard management Roughly what you'd build with Vitess yourself.

Google Spanner was the first public distributed SQL database and popularized the term. Since most of these are proprietary, a vendor lock-in factor exists โ€” and they're not like-for-like swappable (each has design intricacies you hit later), so the advice is to run a POC and a small production feature before committing fully.

09

ACID in NoSQL: possible, but you pay in locks

NoSQL databases can do ACID and strong consistency in distributed transactions โ€” but the same way relational shards do: with locks, which kills on-the-fly horizontal scaling.

Database How it handles distributed ACID
MongoDB Single-document ops are atomic (document = RDBMS row, collection = table). Supports multi-document transactions across sharded clusters & replica sets. In a multi-shard transaction, each shard's commit becomes visible as soon as it's done โ€” even if another shard's commit is still in progress. Uses its own isolation levels; replicates via the Oplog (operation log replayed on secondaries). Cross-shard ACID uses locks, so concurrent writes wait. MongoDB estimates ~90% of apps with the right data model won't need distributed transactions.
AWS DynamoDB ACID guarantees only within the single region where the write originates โ€” not global.
Google Spanner Pessimistic locking, plus 2PC when needed; conflicting concurrent writes wait and retry.
Cassandra No ACID transactions โ€” built for high write frequency with eventual consistency.
โš–๏ธ
The recurring verdict

In a distributed environment, locks are the price of strong consistency โ€” whether it's a sharded MySQL cluster or a NoSQL store. If ACID and strong consistency are the priority, use a relational database; if you can accept eventual consistency, use NoSQL. And when you need both scale and ACID, a polyglot persistence model can bridge the gap.

10

Recap

Two-Phase Commit Prepare โ†’ commit, coordinator + participants, write-ahead logs. Strong consistency but blocking; slowest node and coordinator failure are the weak points.
Saga Sequence of local transactions with idempotent compensations. Non-blocking, developer-owned, eventual-leaning. Choreography (events) or orchestration (controller).
2PC vs. Saga No formula. 2PC for controllable strong consistency; Saga for multi-step flows across independent parties (the bank-transfer test).
Sharded code Route to the right shard; avoid cross-shard transactions. When unavoidable: API Composition (in-memory join) or CQRS (pre-computed query shard).
Real-world scale Dropbox Edgestore (external txn log + copy-on-write, 5โ€“10% cross-shard); Slack & Square on Vitess; VIndex resolves key-less queries.
Off-the-shelf Distributed SQL (Spanner, CockroachDB) is built-to-scale relational; NoSQL ACID (MongoDB, DynamoDB) works but locks limit scaling.
๐Ÿ’ก
The single idea underneath it all

A distributed transaction is expensive no matter how you slice it โ€” lock everyone and block (2PC), or chain undoable steps and risk partial states (Saga). So the best engineers spend most of their effort making the problem rare: co-locate data so one transaction touches one shard, and let a shard manager hide the rest. The cheapest cross-shard transaction is the one you designed away.

11

Further reading

The lessons cite many external links. Below are the ones that resolve live, grouped by topic. The author's own course cross-links and one untested framework link (AxonIQ) are omitted.

Saga frameworks

[1]
Apache Camel Why read it: an integration framework for implementing the Saga pattern in microservices (Section 02).

Sharding at scale โ€” real engineering write-ups

[2]
Cross-shard transactions at 10 million requests per second โ€” Dropbox Why read it: the full account of Section 05 โ€” the external transaction log and copy-on-write tweaks to 2PC.
[3]
(Re)introducing Edgestore โ€” Dropbox Why read it: how Dropbox's shard manager abstracts thousands of MySQL instances from the app.
[4]
Scaling Datastores at Slack with Vitess โ€” Slack Engineering Why read it: the source for Section 06's Slack story โ€” 2.3M QPS, the hotspot problem, and the move to Vitess.
[5]
Remodeling Cash App payments โ€” Square Why read it: Square's entity-group model for keeping money transactions on one shard (Section 06).
[6]
Cross-shard queries with lookup tables โ€” Square Why read it: the VIndex / lookup-table technique behind Section 07's key-less query problem.
[7]
Scaling services with Shard Manager โ€” Meta Engineering Why read it: Facebook's generalized shard manager โ€” the pattern behind Section 05's closing point.
[8]
Vitess Why read it: the open-source shard manager at the center of Sections 06โ€“07.

ACID in NoSQL

[9]
ACID Transactions โ€” MongoDB Why read it: backs Section 09's MongoDB details, including the ~90% "won't need distributed transactions" estimate.
[10]
DynamoDB Transaction APIs โ€” AWS Why read it: confirms Section 09's point that DynamoDB's ACID guarantees are single-region only.