Zero-Downtime Data Migration: How Shopify Solves Hot Shards

A scalable architecture relies on horizontal scaling. For relational databases, this typically translates to horizontal partitioning (sharding). However, sharded architectures inevitably face the hot shard problem: an isolated subset of tenants experiences a massive surge in traffic, consuming disproportionate compute and I/O while sibling shards sit idle.

At Shopify’s scale, an unmitigated hot shard does not just degrade a single merchant—it threatens all co-located tenants sharing that database instance. To solve this, Shopify engineered a mechanism to migrate individual tenant data across independent databases dynamically without customer-perceivable downtime.

---

The Architecture: Pods, Shards, and Multi-Tenancy

Shopify structures its infrastructure around a construct known as a Pod (distinct from a Kubernetes pod):

• Pod Definition: A pod is an autonomous, fully functional, logical grouping of application servers and a shared, dedicated primary database (MySQL).
• Tenant Isolation: Each pod hosts hundreds or thousands of distinct shops. Every relational table contains a discriminator column—typically shop_id (e.g., orders, products, customers).
• No Cross-Pod Shard Overlap: Shards do not share state. Pod 1 and Pod 2 maintain completely isolated MySQL instances.
• Routing Proxy Layer: Front-facing proxy layers (such as NGINX) run custom dynamic routing modules. When a request arrives, the proxy extracts tenant context, inspects a central routing table, and dispatches the request to the pod hosting that shop_id.

                 Incoming HTTP Request
                           │
                           ▼
                 ┌──────────────────┐
                 │   NGINX Proxy    │
                 │ (Routing Module) │
                 └─────────┬────────┘
                           │
              ┌────────────┴────────────┐
              ▼                         ▼
       ┌──────────────┐          ┌──────────────┐
       │    Pod 1     │          │    Pod 2     │
       │ (Shop 1 & 2) │          │ (Shop 3 & 4) │
       └──────┬───────┘          └──────┬───────┘
              ▼                         ▼
       ┌──────────────┐          ┌──────────────┐
       │ MySQL Shard1 │          │ MySQL Shard2 │
       └──────────────┘          └──────────────┘

---

The Problem: Why Move Tenant Data?

When a single merchant on Pod 1 runs a flash sale or receives an unexpected viral spike:

1. The Noisy Neighbor Risk: High read/write throughput saturates CPU, memory, and disk I/O on Shard 1. Hundreds of unrelated merchants on the same database experience elevated latencies or cascade into outage scenarios.
2. Resource Skew: Shard 1 runs at 100% capacity while Shards 2, 3, and 4 operate at 5% capacity.

Application servers are stateless; spinning up more API workers takes seconds. The database, however, is stateful. Migrating gigabytes of tenant records across live database instances while mutations continue requires strict transactional guarantees.

How Tenants Are Selected for Migration

Balancing cannot simply allocate a fixed shop count per shard (e.g., “100 shops per DB”), as shops vary drastically in transactional volume. Instead, data heuristics dictate rebalancing:

• Historical Resource Utilization: Long-term CPU, IOPS, and memory consumption.
• Traffic Profiling: Sustained queries per second (QPS) vs. peak transactional writes.
• Forecasting & Flash Sales: Scheduled promotional events allow proactive migration of a high-load tenant to an isolated, low-utilization pod before the traffic event begins.

---

Hard Constraints of Live Shard Migration

Any data migration engine operating in this environment must meet three criteria:

1. Zero Perceivable Downtime: Read/write availability must remain near continuous throughout the process.
2. Zero Data Loss / Zero Corruption: Exactly-once data integrity. Every row, index, and relational constraint must match the source.
3. Minimal Source Degradation: The migration tool must not saturate the source database’s resources during bulk data extraction.

To fulfill these requirements, Shopify built and open-sourced Ghostferry, an engine written in Go designed to copy data filtered by tenant keys between live MySQL instances.

---

The 3-Phase Migration Process

┌─────────────────────────────────────────────────────────────┐
│ Phase 1: Batch Copy & Binlog Tailing                        │
│ • Record starting binlog coordinates (Point A)              │
│ • Bulk copy rows WHERE shop_id = ? via multi-threaded cursor│
│ • Simultaneously buffer live updates from Binlog >= Point A │
└──────────────────────────────┬──────────────────────────────┘
                               │
                               ▼
┌─────────────────────────────────────────────────────────────┐
│ Phase 2: CDC Replay & Catch-Up                              │
│ • Replay filtered tenant events onto Target DB              │
│ • Monitor replication lag until it drops to 1-3 seconds     │
└──────────────────────────────┬──────────────────────────────┘
                               │
                               ▼
┌─────────────────────────────────────────────────────────────┐
│ Phase 3: Cutover & Atomic Routing Update                    │
│ • Briefly pause writes to Source DB for tenant              │
│ • Drain final CDC delta (Source == Target)                  │
│ • Atomically update routing module at NGINX                 │
│ • Re-enable writes; traffic flows to Target DB              │
└─────────────────────────────────────────────────────────────┘

Phase 1: Batch Copy and Binlog Tailing

Naively running a single snapshot query (SELECT * FROM table WHERE shop_id = X) would lock tables, trigger lock contention, and fail to capture incoming modifications.

1. Pinning Binlog Coordinates: Before extracting data, the migration worker inspects the source database’s Write-Ahead Log (MySQL Binary Log / Binlog) and records the current log file and byte offset ($Pos_0$).

2. Parallelized Batch Iteration: Workers iterate through all tables containing the target shop_id. Using primary key ranges, worker threads query and transfer deterministic chunks (e.g., 500–2,000 rows at a time) and insert them into the target database within transactions.

3. Simultaneous Change Capture (CDC): While the bulk copy proceeds over minutes or hours, live write traffic continues against the source database. Every INSERT, UPDATE, and DELETE on the source appends to the binlog. The migrator tails the binlog starting from $Pos_0$, filtering specifically for entries where the affected row contains the migrating shop_id.

Phase 2: Change Data Capture (CDC) Replay

Once the historical batch copy finishes, the target database contains a base snapshot plus an accumulated delta of live changes that occurred during the transfer:

• The binlog parser extracts row-level changes matching the target shop_id.
• Changes are applied to the target database in commit order.
• If an UPDATE or DELETE arrives for a row that the batch copier already migrated, it updates the target state. If an event occurs for a row not yet reached by the batch copy, Ghostferry handles row collision idempotently (e.g., using INSERT ... ON DUPLICATE KEY UPDATE semantics or tracking keys).
• The migrator tracks replication lag—the time delta between the source commit timestamp and the target applied timestamp.

Phase 3: Cutover and Atomic Routing Update

Once replication lag falls below an acceptable threshold (typically 1 to 3 seconds):

1. Pausing Source Writes: The application temporarily pauses writes for that specific merchant. Rather than presenting an error, the edge proxy or application framework queues or retries inflight requests using exponential backoff.
2. Draining the Final Delta: The migrator drains the remaining binlog stream. Because no new writes occur on the source, the target database achieves absolute parity with the source within 1–2 seconds ($SourceDB = TargetDB$).
3. Updating the Routing Layer: The routing module in NGINX is updated: all incoming requests for shop_id = X are redirected from Pod 1 to Pod 2.
4. Resuming Writes: The write pause is lifted. Inflight retried writes and new requests execute against Pod 2.
5. Validation and Cleanup: Asynchronously, data integrity checks confirm record counts and checksums across both pods. Once verified, a background job purges the migrated tenant’s historical rows from Pod 1 to reclaim disk capacity.

---

Failure Modes and Architectural Trade-Offs

Mechanism / Decision	Trade-Off / Risk	Mitigation Strategy
Filtered Binlog Replay	High CPU overhead parsing every row event across an entire multi-tenant shard.	Use row-based binary logging with optimized Go-based parsers; filter by shop_id before deserializing full payloads.
Dual-Phase Sync (Batch + Tail)	Out-of-order mutations can cause inconsistencies if not tracked accurately.	Strict checkpointing at $Pos_0$; handle idempotency on replay.
Application Write Pause during Cutover	Micro-latency spikes for mutations during the 1-3 second cutover window.	Client-side and middleware retry queues mask the pause, preventing hard 5xx failures.
Post-Migration Purge	Mass DELETE queries on the source DB can cause long table locks and binlog bloat.	Purge data in small batches asynchronously during off-peak hours using primary key ranges.

---

Summary

Dynamic rebalancing is critical to operating shared-nothing relational shards at scale. By combining:

1. Logical tenancy boundaries enforced by discriminator keys (shop_id),
2. Change Data Capture via MySQL binlog streaming,
3. Micro-cutover windows buffered by application retries, and
4. Dynamic proxy routing updates,

large-scale multi-tenant architectures can continuously migrate high-volume workloads and prevent cascading failures without user-facing downtime.