Shard Migration Cutover and the Dual-Write Window

Moving a partition between machines is the operation that makes a partitioned store elastic, and it is also the operation most likely to lose a write. The difficulty is not the bulk transfer; copying a terabyte over a fast link is a solved problem. The difficulty is the interval in which the data exists in two places and both are being written, and the moment at which the system must agree that one of them is authoritative. That interval has a beginning, an end, and a set of invariants that must hold throughout.

Why a copy is not a move

A naive move reads the partition from the source, writes it to the destination, and updates the routing metadata. Every write that arrives during the copy is applied to the source and missed by the destination, so the destination ends up missing everything written after the copy started. The size of the gap is proportional to the copy duration, which for a large partition is minutes. A move that drops a minute of writes is not a move.

Three families of solution exist, and they differ in where the writes go during the window.

Stop the world. Fence writes to the partition, copy it, unpublish the source, publish the destination, resume writes. The data is consistent by construction because there is nothing to reconcile. The cost is the write outage, which equals the copy time. For a small partition on a fast link this is milliseconds and the approach is fine; for a large partition it is unacceptable, and the outage itself causes the kind of retry pile-up that turns a short stall into a long one.

Copy then catch up. Copy a snapshot of the partition while writes continue to the source, recording the write stream or relying on the storage engine’s change history. When the bulk copy finishes, replay the changes since the snapshot onto the destination. The destination is nearly caught up, and the remaining gap is short enough that a brief fence at the end is tolerable. This is the standard approach and it has a specific weakness: the change stream must be retained for the whole copy, so the source needs enough log retention to cover a copy that may take longer than expected.

Write both. Send every write to both source and destination for the duration of the move, with the destination buffering or applying them. The destination is current at all times, and the cutover is a metadata change. The cost is that the source and destination must agree on write order, and the destination must be able to accept writes for a range it does not yet fully own , including writes for keys whose older versions have not arrived yet. Ordering violations produce a destination that has the new value of a key but not the old one, which is harmless for a last-write-wins store and fatal for one that needs history.

In practice the second and third are combined: bulk copy a snapshot, stream the delta, and dual-write only for the final catch-up window when the delta is small.

The state machine

A migration is a sequence of durable states, each of which must be recoverable if the process driving it dies. The states are the contract; a move that skips one is a move that can lose data.

PENDING     destination allocated, nothing copied, source owns everything
COPYING     bulk copy running, source owns reads and writes
CATCHUP     bulk copy done, delta streaming, source still owns reads and writes
DUAL        both accept writes, source owns reads, destination has all data
CUTOVER     destination owns reads and writes, source fenced for this range
CLEANUP     source data dropped, metadata settled

The transition that matters is DUAL → CUTOVER. Before it, the source’s answer is authoritative; after it, the destination’s is. The metadata change that flips ownership must be atomic with respect to routing: any reader that resolves the range during the flip must get one answer or the other, never both, never neither. This is why the ownership record lives in a consensus-backed metadata service rather than in each node’s local view. A single compare-and-set on the range’s owner, conditional on the expected current owner, is the primitive that makes the flip safe; it is the same primitive that underlies Routing Metadata Staleness and the Cost of a Wrong Guess.

The DUAL state exists to make the flip safe in the other direction. If writes go only to the source until the flip, then a write that arrives at the source one microsecond before the flip is applied to the source but not the destination, and the destination becomes authoritative without it. Dual-writing closes that window: by the time ownership flips, every acknowledged write exists in both places. The flip then only changes where reads go, which is a much weaker requirement.

Dual-writing introduces its own hazard. The two copies must converge, and if the source applies write W1 then W2 while the destination receives W2 then W1, a last-write-wins merge at the destination can end with W1’s value. The fix is to attach a version to each write (a log position, a hybrid timestamp, or a monotonic per-key sequence) and have the destination reject or reorder anything older than what it holds. Without such a version, dual-writing is only safe for writes that commute.

Fencing the source

The source must stop serving the range at the flip, and “stop serving” has to be enforced rather than requested. A source that has been told to stop but is still answering reads from a stale local view will return data that the destination has already superseded.

Fencing is usually implemented as a lease or an epoch. The range’s ownership record carries an epoch number, incremented on every ownership change. Every read and write carries the epoch the client believes is current, and a node rejects any request whose epoch is lower than the one it holds. When the flip happens, the epoch increments, and the old source’s epoch is now stale: it rejects the requests it would otherwise have served. A client that receives a rejection refreshes its routing information and retries against the new owner.

The epoch also solves a subtler problem, which is delayed requests. A write sent to the source just before the flip may arrive after it, and if the source has already dropped the range it will be rejected. That is correct: the write never reached the old owner’s durable log, so it was never acknowledged, and the client is entitled to retry. If the write had been acknowledged by the source, it must also have been applied to the destination under the dual-write rule, so the retry against the destination finds the value already there. The invariant is: an acknowledged write exists at the destination before the flip, or it was never acknowledged. Every migration protocol is an attempt to maintain that invariant across the window.

Fencing has a failure mode worth naming. If the metadata service is unavailable at the moment of the flip, the flip cannot be recorded, and the move stalls in DUAL. The source continues to serve reads and both sides continue to accept writes, which is safe but unbounded: the dual-write traffic continues until the metadata service returns. A migration that has been stuck in DUAL for hours is not corrupt, but it is consuming double the write bandwidth and twice the storage for one range. Migration controllers need a timeout that aborts the move by reverting ownership to the source, which is safe precisely because the source never stopped being a valid owner.

What the destination must guarantee before the flip

The flip is only safe if the destination can answer every read the source could answer. That is a stronger condition than “the destination has all the data”, because the destination also needs the local structures that make the data usable.

Secondary structures must be rebuilt, not copied. A destination that receives the base rows but not the local index will answer point reads by scanning, which is correct and arbitrarily slow. Systems that maintain a local index typically rebuild it from the copied rows as part of the copy, which multiplies the copy cost by the index write amplification. Skipping the rebuild is a common source of post-migration latency spikes.

Deletes must be transferred as deletes. If the copy is a logical dump of live rows, rows deleted on the source during the copy never appear on the destination, which is correct. But if the source’s storage uses tombstones with a grace period, the destination must receive the tombstones too, or a delete that has not yet been purged will reappear on the destination as live data. This is the classic resurrection bug, and it is specific to storage engines with delayed purge.

The destination’s clock, epoch, and configuration must be compatible. A destination running an older schema version than the source will reject writes it does not understand, and the rejection arrives during the DUAL window when it is hardest to diagnose. Migrations that cross version boundaries need the schema change to complete before the move starts, not during it.

Cost accounting

The bytes transferred are not the partition size. They are the partition size plus the delta, plus the index rebuild, plus the write amplification of the destination’s storage engine. A partition of size S with a delta of d bytes and an index amplification factor of a moves roughly S·(1+a) + d bytes over the network and writes more than that to disk. The network cost is bounded by the link, so the copy time is at least S·(1+a)/B for bandwidth B, and the fence window at the end is at least d/B plus the time to apply d.

Throttling is not optional. A migration that saturates a node’s network link starves the requests that node is still serving, which raises latency for a range that is not being moved, which causes retries, which adds load. Controllers therefore cap migration bandwidth at a fraction of the link and pause the copy when the source’s request latency exceeds a threshold. The interaction between migration throttling and the cluster’s own rebalancing decisions is where moves go wrong at scale, as described in Rebalance Storms and the Backpressure That Prevents Them.

The time to move one partition also bounds how fast a cluster can respond to a failure. If a node dies holding a hundred partitions, recovery is a hundred concurrent moves, and the aggregate bandwidth is the cluster’s, not one link’s. Recovery time therefore scales with the total data on the failed node divided by the aggregate spare bandwidth, and no amount of per-move tuning changes that.

The alternative: don’t move

Every migration is a chance to lose data and a guaranteed consumption of bandwidth. Systems that can avoid moving data are better off, and there are three ways to do it.

Replicate the data to the new location and switch reads, leaving the old copy as a replica rather than deleting it. The move becomes a replica addition, which is the same protocol but never needs a destructive cleanup step.

Split instead of move. If the reason for the migration is that the partition is too large, splitting it in place leaves half the data where it is and moves nothing. Only the half that exceeds the node’s capacity needs to move, which halves the transfer. The split path is described in Range Partition Boundaries and the Split Key Contract.

Assign at creation. If the partitioning function is flexible enough that a new node can take a share of future keys without taking existing ones, the expensive move is deferred indefinitely. This works only when the key distribution is stable enough that new keys land on the new node, which is not true for ordered keys and is the reason the append-only case is hard.