hard Online Resharding & Cross-Shard Operations
Change the shard count without ever losing a write
Resharding online means moving keys to new shards while the system keeps serving reads and writes — and the whole difficulty is one invariant: there must never be a moment where a key that was written is unreadable. If you bulk-copy rows to a new shard and then flip reads over, every write that arrived during the copy is stranded on the old shard the readers have abandoned. The safe pattern makes the two locations overlap in time (dual-write) so every phase can still serve a correct read, then cuts over atomically per key. Skip the overlap and you get a split-brain of data: the write exists, but nobody reads the copy that has it.
The safe pattern: expand / contract
This is the same discipline as an expand/contract schema migration — add the new path, run both, verify, switch, remove the old — applied to data placement:
- Add shards + start backfill. Stand up the new shards and bulk-copy existing rows to their new home. Reads and writes still go to the old shards.
- Dual-write. The application writes each affected key to both old and new locations. This is the load-bearing step: it guarantees nothing written after the backfill snapshot is lost. Reads still serve from old.
- Verify. Compare old and new — row counts, then checksums per key range — and reconcile any divergence before trusting the new copy. (Backfill races dual-write at the boundary; verification catches the gaps.)
- Cutover. Flip reads to the new shards, atomically per key or per range. Writes stay dual until reads are fully migrated, so a mid-cutover read of either home is correct.
- Stop dual-write + decommission. Once all reads are on new and verified, drop the old writes and retire the old shards.
Traced example
Split hot shard S2 (keys 66–99) into S2a (66–82) and S2b (83–99).
| t | event | where it goes | a read of key 90 sees |
|---|---|---|---|
| t0 | backfill copies existing 66–99 into S2a/S2b | read: S2 · write: S2 | S2 (correct) |
| t1 | dual-write on; write key 90 = X | write to S2 and S2b | S2 (has X) |
| t2 | verify: S2b matches S2 for 83–99 | read: S2 · write: S2 + S2b | S2 (has X) |
| t3 | cutover reads for 83–99 to S2b | read: S2b · write: S2 + S2b | S2b (has X — dual-write saved it) |
| t4 | stop dual-write, drop S2 | read/write: S2a / S2b | S2b |
Now delete step t1's dual-write: the write of key 90 lands only on old S2, but at t3 readers switch to S2b, which never received it. The read returns the pre-write value — a silent lost update. Dual-write is precisely what closes that window.
Why consistent hashing changes the calculus
The migration above is expensive because splitting a range touches a whole interval. On a consistent-hashing ring with vnodes, adding a node reassigns only its arcs (~K/N of keys), so the "keys that must move" set is small and incremental — you still dual-write those keys, but the blast radius per membership change is bounded rather than cluster-wide. Naive hash mod N is the opposite: changing N moves ~80% of keys, turning every resize into a full-cluster dual-write marathon.
Cross-shard operations — the two that break single-node assumptions
1. A transaction spanning shards has no single-node ACID
Move $100 from account A (on S1) to account B (on S3). One node cannot make both changes atomically, so you need a distributed protocol:
- Two-phase commit (2PC). A coordinator sends prepare to S1 (debit) and S3 (credit); each locks the rows and votes. Only if both vote yes does the coordinator send commit; otherwise abort. Strong atomicity, but participants hold locks across the round trip and block if the coordinator dies mid-commit (coordinator is a SPOF) — it is a synchronous, availability-reducing protocol.
- Saga. Do it as local transactions with compensations: debit A (local, committed), then credit B (local); if the credit fails, run a compensating credit-back to A. Non-blocking and highly available, but there is a window where money has left A and not arrived at B (no isolation), and every step needs an idempotent compensating action.
2. A secondary index that doesn't include the shard key forces scatter-gather
Users are sharded by user_id. The query "find the user with email = x" filters on a field that is not the shard key, so the router cannot know which shard holds it and must ask all N shards and merge (scatter-gather): cost scales with N, tail latency is the slowest shard, and effective throughput is divided by N. The fix is a global secondary index sharded by email that returns the (user_id, shard) in one hop — at the cost of maintaining a second distributed structure kept in sync with the base table (async = eventually consistent/stale reads; sync = write amplification and cross-shard write coupling). A local index (indexed within each shard, on the shard key path) avoids that cost but only answers queries that already carry the shard key.
Pitfalls
- Cutover before verify. Flipping reads before checksums pass exposes backfill gaps as missing rows. Verify is not optional.
- Non-idempotent dual-write. Retried writes during migration double-apply unless writes are idempotent (upsert by key, not blind insert/increment).
- Forgetting to stop dual-write. Leaving both paths live permanently doubles write cost and re-creates drift; contract is a required phase, not a nicety.
- 2PC coordinator as a hidden SPOF. A coordinator crash between prepare and commit leaves participants locked and blocked. Log decisions durably and have a recovery path, or prefer a saga.
- Scatter-gather creeping into hot paths. A single non-shard-key query on the critical path caps throughput at 1/N. Catch it in design, not in production.
- Global index consistency assumed. An async global index can lag the base table; a read-after-write on the indexed field may miss. Know whether your index is sync or async.
Selection & trade-offs — how a senior engineer decides
Dual-write cutover vs consistent-hashing incremental. Range-split migrations move whole intervals and need a deliberate backfill→dual-write→cutover campaign per split — use it when you shard by range or use a directory and resizes are rare, planned events. Consistent hashing with vnodes moves only ~K/N keys per membership change, so growth is cheap and frequent — prefer it when the cluster scales elastically and often. You gain incrementalism; you give up range scans and accept ring-management complexity.
Local (shard-key) index vs global index. Default to a local index — zero extra infrastructure — when every query can carry the shard key. Add a global index only for a genuinely hot query on a different field, accepting that you now run and keep-consistent a second distributed structure. The trade is one fan-out query vs perpetual index-maintenance cost.
When a cross-shard transaction is worth 2PC/saga vs denormalizing to avoid it. A distributed transaction is expensive (2PC: blocking, locks, coordinator SPOF; saga: no isolation, compensations to write and test). If the two entities are almost always changed together, the senior move is often to denormalize so they live on the same shard (co-locate by a shared shard key), turning the cross-shard txn into a single-node one. Reserve 2PC for genuine cross-domain atomicity you cannot co-locate (money movement across independently-owned ledgers), and reach for a saga when you need availability and can tolerate a visible intermediate state with compensations.
Takeaways
- Online resharding is safe only if the old and new homes overlap in time: backfill → dual-write → verify → cutover → stop dual-write. Skipping dual-write strands writes made during migration — a split-brain of data.
- Consistent hashing/vnodes bound the keys that move to ~K/N per change, making resize incremental; naive
hash mod Nmoves ~80% and turns every resize into a full migration. - A cross-shard transaction has no single-node ACID: use 2PC (atomic but blocking, coordinator SPOF) or a saga (available but no isolation, needs idempotent compensations) — or denormalize to co-locate and avoid it.
- A query on a non-shard-key fans out to all shards (scatter-gather); a global secondary index localizes it to one shard, at the cost of keeping a second distributed structure consistent.
Re-authored for this guide; both diagrams hand-authored as SVG. Sources: Martin Kleppmann, Designing Data-Intensive Applications, ch. 6 & 7 (Partitioning, Distributed Transactions); the expand/contract migration pattern (Sato/ThoughtWorks; Stripe and GitHub engineering write-ups on online resharding); Gray & Reuter on two-phase commit; Garcia-Molina & Salem, Sagas (1987). See also: Sharding Strategies, Saga, Transactional Outbox, Consistency Spectrum, Idempotency.
🤖 Don't fully get this? Learn it with Claude
Stuck on Online Resharding & Cross-Shard Operations? Open Claude, copy a block below, and it'll teach you this exact concept — visually and interactively.
Build the mental picture, not memorization.
I just read a lesson on **Online Resharding & Cross-Shard Operations** (System Design) and want to truly understand it. Explain Online Resharding & Cross-Shard Operations from first principles using ONE vivid real-world analogy and a visual mental model — draw it as ASCII art or a clear step-by-step diagram — with a concrete example using real numbers. Then ask me one question to check I got the mental picture, and wait for my reply. If you're unsure or a claim isn't standard, say so and reason from first principles instead of guessing.
Socratic — adapts to where you're stuck.
Teach me **Online Resharding & Cross-Shard Operations** interactively. Ask me ONE guiding question at a time, wait for my answer, and adapt to my confusion — build the idea with me step by step instead of explaining it all at once. If you're unsure or a claim isn't standard, say so and reason from first principles instead of guessing.
Active recall exposes what you missed.
Quiz me on **Online Resharding & Cross-Shard Operations** with 5 questions, easy to tricky, ONE at a time. Tell me if each answer is right; at the end, explain clearly what I got wrong and why. If you're unsure or a claim isn't standard, say so and reason from first principles instead of guessing.
Intuition + hook + flashcards for long-term memory.
Help me remember **Online Resharding & Cross-Shard Operations** for the long term: give the one-sentence intuition, a memorable hook/mnemonic, a tiny worked example, and 3 active-recall flashcards (Q -> A). If you're unsure or a claim isn't standard, say so and reason from first principles instead of guessing.