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.
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.
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.
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:
Create order
โ compensate: cancel order.
Process payment
(from card info) โ compensate: refund.
Update inventory
โ compensate: restock.
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.
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.
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 CompositionAn 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
External transaction log.
The 2PC coordinator writes all protocol events to a separate log. Even if every shard commits and acks, the
transaction isn't considered committed until the log records it. So any concurrent read/write only checks the
log
to learn transaction state, instead of contacting each shard โ cutting lock contention and
distributed-deadlock risk.
Copy-on-write at commit staging.
In the staging phase (participants ready but not committed), changes are written via copy-on-write to a
dedicated MySQL table, not the primary tables; only at final commit does the data move to the primary tables. This "shadowing"
handles concurrent writes efficiently โ like a snapshot in snapshot isolation.
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.
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 CommitPrepare โ commit, coordinator + participants, write-ahead logs. Strong consistency but blocking; slowest
node and coordinator failure are the weak points.
SagaSequence of local transactions with idempotent compensations. Non-blocking, developer-owned,
eventual-leaning. Choreography (events) or orchestration (controller).
2PC vs. SagaNo formula. 2PC for controllable strong consistency; Saga for multi-step flows across independent
parties (the bank-transfer test).
Sharded codeRoute to the right shard; avoid cross-shard transactions. When unavoidable: API Composition (in-memory
join) or CQRS (pre-computed query shard).
Off-the-shelfDistributed 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 CamelWhy read it: an integration framework for implementing the Saga pattern in microservices
(Section 02).