system-design · intermediate

Rebalancing Shards Under Skewed Traffic

The Central Question

Consider a sharded multi-tenant database platform running on the DataLab platform (datalab.com) processing 100,000,000 queries per day:

• The platform distributes tenant records across 4 database shards (Shard 0, Shard 1, Shard 2, Shard 3).
• Over 6 months, a newly onboarded enterprise customer (Tenant #901 assigned to Shard 1) grows rapidly, producing 60% of all platform transactions.
• Shard 1 experiences 98% CPU saturation, 95% disk usage, and continuous query timeouts, while Shard 0, Shard 2, and Shard 3 sit idle at 15% CPU and 20% disk usage.

If the engineering team attempts to rebalance Shard 1 by taking the database offline for 8 hours to manually copy tables, the company violates its 99.99% SLA, incurring massive downtime losses.

Shard Rebalancing is the automated operational process of migrating partition subsets or hash slots from over-burdened database nodes to under-utilized or newly added server instances in real time with Zero Application Downtime.

This lesson answers one central question: How do distributed databases execute zero-downtime live shard rebalancing using 4-stage migration pipelines (Copy $\rightarrow$ Dual-Write $\rightarrow$ Catch Up $\rightarrow$ Shadow Cutover) and Range Splitting algorithms while maintaining ACID consistency?

Causes of Traffic Skew and Shard Imbalance

Shard imbalance occurs when data size or QPS distributes non-uniformly across physical database nodes:

flowchart TD
  SkewCauses[Root Causes of Shard Skew] --> HotTenant[1. Hot Tenant Concentration]
  SkewCauses --> AutoInc[2. Sequential Range Monotonic Writing]
  SkewCauses --> Hardware[3. Heterogeneous Server Hardware]
  
  HotTenant --> HTDesc["One massive customer (Tenant 901) produces 60% of QPS.<br/>All data lands on Shard 1."]
  AutoInc --> AIDesc["Range sharding by date or auto-increment ID.<br/>100% of today's writes hit the newest Shard."]
  Hardware --> HWDesc["Mixing older 16-core servers with new 64-core servers.<br/>Identical slot allocations cause older nodes to collapse."]

Figure 1: Taxonomy of root causes triggering shard traffic skew and data imbalance.

The 4-Stage Zero-Downtime Live Migration Workflow

To move a partition or tenant dataset from a source node (Shard 1) to a target destination node (Shard 4) without taking the application offline, databases execute a 4-Stage Live Migration Pipeline:

sequenceDiagram
    autonumber
    actor Router as Database Proxy / Router
    participant Src as Source Shard 1 (Overloaded)
    participant Dst as Target Shard 4 (New)
    participant CDC as Change Data Capture (CDC Stream)
    
    Note over Src,Dst: Stage 1: Bulk Snapshot Copy
    Src->>Dst: 1. Stream Initial Table Snapshot (Bulk Copy)
    
    Note over Src,Dst: Stage 2: Dual-Writing & CDC Replication
    Router->>Src: 2. Live Writes (INSERT/UPDATE)
    Src->>CDC: 3. Log WAL Mutation to CDC Stream
    CDC->>Dst: 4. Replicate Real-Time Delta Writes
    
    Note over Src,Dst: Stage 3: Catch-Up Verification
    Dst-->>Router: 5. Replication Lag Drops to < 1ms (Caught Up!)
    
    Note over Src,Dst: Stage 4: Atomic Cutover
    Router->>Router: 6. Switch Router Slot Table to Point to Target Shard 4!
    Router->>Src: 7. Delete Migrated Data (Purge Old Range)

Figure 2: Sequence diagram detailing the 4-stage zero-downtime live shard migration workflow.

Rebalancing Strategies: Dynamic Range Splitting vs. Hash Slot Migration

How data partitions are divided determines how cleanly shards rebalance:

flowchart TD
  RebalanceStrategies[Shard Rebalancing Algorithms] --> RangeSplit[1. Dynamic Range Splitting]
  RebalanceStrategies --> SlotMigrate[2. Fixed Hash Slot Migration (Redis/Citus)]
  
  RangeSplit --> RSDesc["Key Range [1 .. 100,000] becomes too large.<br/>Database splits range into [1 .. 50,000] and [50,001 .. 100,000].<br/>Move second range to new node."]
  SlotMigrate --> SMDesc["Fixed 16,384 slots.<br/>Node A transfers slots 4,000 to 5,460 to Node D.<br/>Zero key re-hashing required!"]

Figure 3: Comparison of Dynamic Range Splitting vs Fixed Hash Slot Migration.

1. Dynamic Range Splitting (CockroachDB / HBase)

• Data is stored in contiguous key ranges called Ranges or Tablets (e.g. 64 MB size cap).
• When a Range exceeds 64 MB, the storage engine automatically splits the range into two equal halves ($32\text{ MB} + 32\text{ MB}$) and migrates one half to an under-utilized node in the background.

Hot Range Load Shedding

Rate-Limiting Live Migration I/O

• Migration background threads are allocated a strict, configurable token bucket bandwidth cap (e.g. max_migration_rate = 20 MB/s).
• If production query latency exceeds $10\text{ms}$, the throttling controller automatically scales back background migration copy threads to 5 MB/s, prioritizing live user transactions over rebalancing speed.

Automated Rollback Procedures for Failed Migration Runs

1. The orchestrator immediately cancels the bulk copy worker threads and drops the incomplete table partitions on the target node.
2. The database router routing table remains unchanged, continuing $100\%$ of production traffic to the source shard without data loss or user disruption.
3. Operating teams receive an alert detailing the migration failure cause before initiating a second rebalance attempt.

Post-Cutover Garbage Collection & Disk Compaction

2. Fixed Hash Slot Migration (Redis Cluster / Vitess)

• The key space is divided into fixed virtual slots (e.g. 16,384 slots).
• Rebalancing consists of moving specific slot ownership entries (e.g. slots 4,000 to 5,000) from Node A to Node D. Clients receive MOVED 4000 192.168.1.50 redirects during migration.

Virtual Node Migration in Consistent Hashing

1. The cluster orchestrator selects a subset of Node A's virtual nodes (e.g. 50 virtual nodes).
2. The orchestrator re-assigns those 50 virtual nodes to a newly provisioned physical Node E.
3. Because virtual nodes are scattered randomly across the ring, rebalancing transfers fractions of data from multiple key ranges uniformly, avoiding large contiguous data moves.

Shadow Cutover and Data Reconciliation Verification

• Dual-Read Shadow Verification: The router sends production SELECT queries to both the source shard and target shard in parallel.
• Data Reconciliation Audit: A comparison worker verifies that returned row byte payloads and cryptographic hashes match $100\%$ between source and target shards.
• Fail-Fast Rollback: If a single row discrepancy or hash mismatch is detected, the router aborts the migration automatically, preserving source shard routing without user impact.

Complete Worked Example: Production Go Live Shard Rebalancing Migration Engine

Let's inspect a complete Go implementation of a Live Shard Rebalancing Migration Engine for the DataLab platform (datalab.com).

package main
import (
	"context"
	"fmt"
	"sync"
	"time"
)
type MigrationState int
const (
	StateIdle MigrationState = iota
	StateBulkCopy
	StateDualWriteCatchUp
	StateCutoverComplete
)
type ShardMigrationEngine struct {
	mu           sync.Mutex
	sourceShard  string
	targetShard  string
	state        MigrationState
	cdcStream    chan string
	migratedKeys map[string]string
}
func NewShardMigrationEngine(source, target string) *ShardMigrationEngine {
	return &ShardMigrationEngine{
		sourceShard:  source,
		targetShard:  target,
		state:        StateIdle,
		cdcStream:    make(chan string, 1000),
		migratedKeys: make(map[string]string),
	}
}
func (e *ShardMigrationEngine) ExecuteLiveMigration(ctx context.Context, keysToMigrate map[string]string) error {
	e.mu.Lock()
	e.state = StateBulkCopy
	fmt.Printf("[STAGE 1: BULK COPY] Starting snapshot migration from %s to %s...\n", e.sourceShard, e.targetShard)
	e.mu.Unlock()
// 1. Bulk Copy Snapshot Data
	for k, v := range keysToMigrate {
		e.mu.Lock()
		e.migratedKeys[k] = v
		e.mu.Unlock()
		time.Sleep(10 * time.Millisecond) // Simulate network bulk transfer
	}
e.mu.Lock()
	e.state = StateDualWriteCatchUp
	fmt.Printf("[STAGE 2: DUAL WRITE CATCH-UP] Bulk copy finished. Consuming CDC delta stream...\n")
	e.mu.Unlock()
// 2. Consume CDC delta mutations until replication lag is 0
	e.flushCDCStream()
// 3. Atomic Cutover
	e.mu.Lock()
	e.state = StateCutoverComplete
	fmt.Printf("[STAGE 3: ATOMIC CUTOVER] Router updated! All traffic for migrated keys now routes to %s.\n", e.targetShard)
	e.mu.Unlock()
return nil
}
func (e *ShardMigrationEngine) RecordCDCMutation(key string, value string) {
	e.mu.Lock()
	defer e.mu.Unlock()
if e.state == StateDualWriteCatchUp || e.state == StateBulkCopy {
		e.cdcStream <- fmt.Sprintf("%s:%s", key, value)
	}
}
func (e *ShardMigrationEngine) flushCDCStream() {
	for len(e.cdcStream) > 0 {
		mutation := <-e.cdcStream
		fmt.Printf("[CDC REPLICATION] Applying real-time delta write to Target: %s\n", mutation)
	}
}

Failure Modes and Engineering Mitigations

What You Should Remember

1. Rebalancing redistributes skewed partitions: Move data subsets away from overloaded nodes to under-utilized or newly added server instances.
2. Execute zero-downtime 4-stage migrations: Bulk Copy $\rightarrow$ Dual-Write CDC $\rightarrow$ Catch-Up $\rightarrow$ Atomic Cutover.
3. Throttle migration network bandwidth: Limit bulk copy transfer speeds to prevent migration traffic from degrading live application user queries.
4. Use Fixed Hash Slots or Dynamic Ranges: Fixed slots (16,384 in Redis) and 64 MB Range splits avoid global data re-hashing when adding nodes.
5. Verify data integrity before atomic cutover: Ensure CDC replication lag is $<1\text{ms}$ and row checksums match before updating router tables.

Glossary of Terms

Practice Scenario and Self-Assessment

Architecture Scenario

• Shard 2 reaches 95% disk capacity due to a sudden merchant promotion.
• A new server instance (Shard 5) has been provisioned to relieve Shard 2.

1. Detail the 4-stage live migration plan to move 500 GB of merchant data from Shard 2 to Shard 5 with zero downtime.
2. Formulate the bandwidth throttling and CDC catch-up verification strategy.

Interactive Self-Assessment

It streams real-time database mutations occurring during bulk copy, ensuring the target shard is up-to-date before cutover.

CDC automatically converts SQL integer primary keys into string UUIDs.

CDC replaces public DNS nameservers with local hosts file entries.

CDC doubles the physical hardware clock speed of target database CPUs.

The bulk migration copy consumes 100% of the source node's network and disk I/O, starving live user queries and causing outages.

Un-throttled migration automatically formats persistent NVMe SSD disk drives on target nodes.

Un-throttled migration revokes edge HTTPS TLS encryption certificates on load balancers.

Un-throttled migration reboots operating system hypervisors across all shard nodes.

What to Learn Next

• Hot Partition / Hot Key — Mitigating Skewed Traffic: Master key salting and local micro-caching.
• Database Sharding — Split Data Across Many Machines: Revisit shard keys and hash partitioning.
• Data Replication — Single-Leader, Multi-Leader, and Leaderless: Review replication lag and consistency models.

Track: Data, Storage and Messaging

Previous: How to Scale a Database — The Progressive Scaling Ladder

Next: Search Index Freshness vs Ranking Quality

By Shubham Jain

All articles · Study paths

Shubham Jain · Learning Lab