Deciding Where Shops Should Live

Shopify's MySQL sharding model attaches a shop_id to shop-owned tables, and a shard hosts many shops. Since shop sizes vary widely, simply counting shops per shard doesn't reflect true utilization. An earlier approach classified shards by historical database utilization and traffic patterns into cohorts like high_traffic and low_traffic, then migrated shops between cohorts using a scheme such as moving every Nth shop from a hot shard to a cooler one. Those proposed moves were simulated and validated against forecasted effects before execution.

That strategy works, but it's not the only one. Placement logic can be arbitrarily complex, prioritizing different metrics such as shop size, gross merchandise value, migration time, or flash-sale patterns. Typically, multiple placement hypotheses are tested against recent data before an ideal shop distribution is chosen. Once determined, the system generates a list of moves to achieve the target state.

Constraints on the Move Process

Moving a shop from one shard to another—say, from Pod 1 to Pod 2—hinges on three hard requirements.

  • Availability: The move must happen entirely online. The merchant's storefront stays live while data is copied, with no visible downtime for buyers or staff.
  • Data integrity: No data can be lost or corrupted during the migration. All data present at move start must be copied, and every write that hits the source after the move begins must also be replicated to the target.
  • Throughput: The move must complete in a timely fashion. Shops range widely in size, and running many migrations in parallel shouldn't strain the infrastructure.

A move breaks down into three phases: (1) batch copying and binlog tailing, (2) cutover, and (3) control-plane update, traffic serving, and stale-data pruning.

Phase One: Batch Copy and Binlog Tailing

For the data migration itself, Shopify uses Ghostferry, its in-house open-source library originally built for cloud migration, designed to copy data between MySQL instances. Ghostferry relies on two mechanisms: batch copying of existing rows and tailing MySQL's binlog for ongoing changes.

Batch copying iterates over the source's tables, selects rows matching the target shop's ID, and writes them to the target shard, with each batch committed in its own MySQL transaction. To avoid race conditions where source rows change mid-copy, Ghostferry uses MySQL's SELECT...FOR UPDATE locking reads. This write-locks the selected rows for the transaction duration, guaranteeing the read-then-write transaction stays atomic and the source data remains unchanged until committed on the target.

In parallel, Ghostferry consumes MySQL's binlog. With row-based replication, the binlog records every row-level operation, serving as the source of truth for all changes. Ghostferry streams those events, filters for ones relevant to the shop being moved, and replays them on the target. For throughput, copying runs concurrently across multiple tables in separate threads. Throughout this phase, the merchant's storefront stays fully active on Pod 1.

Phase Two: Cutover

Once batch copying finishes, all pre-existing data for the shop now resides on Pod 2. Ghostferry keeps replaying new binlog events until the queue of pending events is nearly drained—effectively real-time, on the order of seconds. At that point, Ghostferry enters cutover.

During cutover, writes to the source must stop so no new binlog events are generated. Ghostferry records the source's final binlog position as the stopping coordinate, processes the remaining queued events, and marks copying complete upon reaching that position.

Cutover safety is enforced at the application level with a multi-reader-single-writer (MRSW) lock backed by Redis. Before a move begins, any unit of work—web request or job—scoped to the shop must hold the reader (shared) portion of the lock; multiple readers are fine as long as no writer holds the exclusive side. Jobs that can't be scoped to a single shop must hold a similar global lock. When entering cutover, the shop mover waits for reader locks to release, then acquires the writer (exclusive) lock, asserting no shop writes execute on that pod. If the exclusive lock can't be acquired in time, the move fails rather than risking data loss.

Phase Three: Route Update, Traffic, and Cleanup

With data consistency guaranteed, the shop mover updates the control plane's routing table—stored in a separate, unsharded database—to point the shop to its new pod. Once that update lands, the mover releases the exclusive lock, letting work proceed and writes resume on Pod 2. The cutover window, designed to be short, is the only opportunity for downtime in the entire process.

The shop now serves from its new location, though Pod 1 still holds a copy of its data. Before pruning, the system verifies the move succeeded and confirms no queries are routed back to the source shard for a period after migration. Once deemed successful, stale data on the old pod is cleaned up.

Both high-traffic merchants are now isolated at the pod level, database utilization is smoothed across the platform, and Peppy Peppers operates entirely from Pod 2.

Verification: The Safety Net for Online Migrations

Moving data between live MySQL shards is inherently risky. Ghostferry’s phases—batch copying, binlog tailing, cutover, and control-plane updates—only get you partway there. The harder problem is proving that the migration actually did what it was supposed to, especially as the tool gains concurrency support and the ability to interrupt and resume mid-flight.

Shopify’s answer is a layered verification strategy. A suite of verifiers runs before, during, and after every migration, checking for data corruption, ensuring completeness of the transfer, and confirming authenticity of the copied rows. This isn’t just defensive coding; it’s a core design principle for the entire system.

Beyond runtime checks, the core algorithm behind Ghostferry is also formally specified in TLA+. Modeling the algorithm in TLA+ lets the team reason about its correctness at a logical level, catching design flaws that runtime testing might miss.

This degree of rigor matters because the stakes are concrete. Moving a shop from one shard to another touches large, interconnected systems that merchants depend on daily. The ability to rebalance shards with confidence lets Shopify keep infrastructure stable and well-provisioned, even as data grows into terabytes and shops scale.

Why Shard Rebalancing Matters

The practical benefit of a reliable migration pipeline is operational agility. Shard rebalancing is the mechanism Shopify uses to keep its database tier healthy. When one shard gets hot or approaches capacity, being able to move shops away prevents performance degradation across the board. Without a tool like Ghostferry, a rebalance would require downtime and manual intervention; with it, the process is automated and can run without impacting production traffic.

The need for this capability only grows as Shopify’s platform expands. Store owners build their businesses on the infrastructure, and any disruption to the underlying database has direct consequences. High-confidence shard rebalancing is a foundational capability for maintaining the platform’s reliability at scale.

For engineers interested in this kind of work, Shopify continues to invest in the database engineering space, looking for people who want to build autonomous tooling for data placement and organization.