Cross-Shard Operations & Distributed Transactions

Sharding breaks joins and atomic multi-key writes

Sharding buys scale by breaking two things you took for granted on a single database: joins and atomic multi-key writes. Once related rows can live on different nodes, a query that spans them is a distributed operation, and a write that must change both is a distributed transaction.

Cross-shard reads (scatter-gather)

A query that is not scoped to one shard key must fan out to every partition, and it is bounded by the slowest shard, not the average. This is tail latency amplification: if each shard's p99 is 10ms and you hit 50 shards, the chance that at least one is slow is 1 minus 0.99^50, roughly 40%, so the overall p99 is far worse than 10ms. Mitigations: avoid the fan-out by choosing the shard key to match the query, denormalize so the data you need is co-located, cap the fan-out width, and use hedged/speculative requests to blunt single-shard tail latency. The senior instinct is to design most reads to touch one shard and treat scatter-gather as the rare, budgeted case.

Cross-shard writes: avoid 2PC on the hot path

The textbook answer is two-phase commit (2PC): a coordinator asks all participants to prepare (lock and promise), and if all vote yes, tells them to commit. It gives atomicity, and you should know it, but avoid it on the hot path. 2PC holds locks across a network round trip, so it kills throughput, and it is blocking: if the coordinator dies after prepare, participants sit holding locks indefinitely, unsure whether to commit or abort. It is defensible only for low-frequency, must-be-atomic operations, and modern systems (Spanner, CockroachDB) make it viable only by pairing it with tight clock/consensus machinery you do not want to hand-roll.

Sagas: the standard replacement

A saga breaks the operation into a sequence of local transactions, each on one shard, and for every step defines a compensating action that semantically undoes it. If a later step fails, you run the compensations for the completed steps in reverse. A money transfer becomes: debit A (local), credit B (local); if credit fails, compensate by re-crediting A. You give up isolation (intermediate states are visible, so you design for them, e.g. a "pending" balance) in exchange for no distributed locks and independent, available shards. Sagas are orchestrated (a central coordinator drives steps, easier to reason about and monitor) or choreographed (each step emits an event the next reacts to, more decoupled but harder to trace).

Two patterns make sagas safe under real-world at-least-once delivery:

• Idempotency keys. Every step carries a unique key; the receiving shard records processed keys and dedups retries, so replaying "credit B" after a timeout does not double-credit.
• The outbox pattern. The problem: you must both write the DB row and publish an event, but they are separate systems, so a crash between them loses one. The fix: in the same local transaction, write the business row and an "outbox" row; a separate relay reads the outbox and publishes to Kafka, marking rows sent. Now the DB write and the intent-to-publish are atomic, and the relay retries publishing idempotently. This is how you drive a saga's next step reliably.

Interview nuance: the failure mode interviewers hunt for is hand-waving cross-shard joins and multi-key writes as if they were free. When the design crosses shards, say it: "this is now a distributed transaction; I will use a saga with compensations and idempotency keys, not 2PC on the hot path, and I will denormalize to keep the frequent reads single-shard."

Recap: sharding breaks joins (scatter-gather, bounded by the slowest shard) and atomic multi-key writes; avoid 2PC on the hot path because it locks and blocks on coordinator failure; use a saga of local transactions with compensating actions, make retries safe with idempotency keys, publish events atomically with the outbox pattern, and denormalize to avoid cross-shard joins.