Rebalancing Shards After Adding Nodes to a Consistent Hash Ring

This walkthrough takes a PostgreSQL-backed service that routes keys through a consistent hash ring with virtual nodes and adds two new database nodes (db4, db5) — computing exactly which vnode ranges move, pre-copying them in throttled batches, verifying with dual reads, flipping the ring version atomically in etcd, and deleting the moved rows. It applies the general procedure from shard rebalancing & data movement, part of the Shard Migration & Rebalancing Operations guide.

The starting point: three PostgreSQL 15 nodes (db1–db3), 64 vnodes each, an events table of ~900 GB total, and application routing built along the lines of the step-by-step consistent hash routing guide. Each row stores its precomputed ring_hash (the 32-bit hash of its key), which makes moved ranges directly queryable.

Prerequisites

Step 1 — Compute which vnode ranges move

On a consistent hash ring, adding nodes moves only the ranges that now fall between a new node’s vnodes and their predecessors — roughly 2/5 of the keyspace when going from three nodes to five, and nothing else. Compute the exact move set by building the old and new rings and diffing ownership:

import hashlib
from bisect import bisect_right

VNODES, RING = 64, 2**32

def build_ring(nodes):
    ring = []
    for node in nodes:
        for i in range(VNODES):
            h = int(hashlib.md5(f"{node}:vn{i}".encode()).hexdigest(), 16) % RING
            ring.append((h, node))
    return sorted(ring)

def owner(ring, h):
    keys = [p for p, _ in ring]
    return ring[bisect_right(keys, h) % len(ring)][1]

old_ring = build_ring(["db1", "db2", "db3"])
new_ring = build_ring(["db1", "db2", "db3", "db4", "db5"])

moves = []
for i, (h, node) in enumerate(new_ring):
    if node in ("db4", "db5"):                      # only new vnodes gain ranges
        lo = new_ring[i - 1][0]                     # predecessor on the new ring
        moves.append({"lo": lo, "hi": h,
                      "src": owner(old_ring, h), "dst": node})

moved = sum((m["hi"] - m["lo"]) % RING for m in moves)
print(f"{len(moves)} ranges, {moved / RING:.1%} of keyspace moves")
for m in moves[:3]:
    print(f"  ({m['lo']:>10}, {m['hi']:>10}]  {m['src']} -> {m['dst']}")

Running this prints 128 ranges (64 per new node) totalling about 40% of the keyspace — matching the theoretical 2/5 — with each range naming its exact source and destination. Persist moves as JSON; every later step (copy, verify, delete) must consume this file rather than recomputing, so all phases agree on boundaries.

Operational note: The hash function and vnode naming (f"{node}:vn{i}") must byte-for-byte match what the application router uses. A planner using md5 while the router uses crc32 produces a move set that is confidently, silently wrong — data lands where no router will ever look for it.

SRE tip: Sanity-check the plan before touching data: if the printed percentage differs from new_nodes / total_nodes by more than a couple of points, or any range’s src is a node not in the old ring, stop. Also run a dry-run count per range (SELECT COUNT(*) FROM events WHERE ring_hash > lo AND ring_hash <= hi) on the sources to estimate copy duration against your maintenance window.

Step 2 — Pre-copy the moved ranges in throttled batches

With db4 and db5 taking no traffic yet, bulk-copy each range from its source using keyset pagination, version-guarded upserts, and a rate cap so db1–db3 keep serving production comfortably:

import json, time, psycopg2

BATCH, MAX_ROWS_PER_SEC = 5000, 20000
moves = json.load(open("moves.json"))

def copy_range(m, conns):
    src, dst = conns[m["src"]], conns[m["dst"]]
    last_id = m.get("resume_pk", 0)
    while True:
        t0 = time.monotonic()
        with src.cursor() as s:
            s.execute("""SELECT id, ring_hash, payload, row_version
                         FROM events
                         WHERE ring_hash > %s AND ring_hash <= %s AND id > %s
                         ORDER BY id LIMIT %s""",
                      (m["lo"], m["hi"], last_id, BATCH))
            rows = s.fetchall()
        if not rows:
            return
        with dst.cursor() as d:
            d.executemany("""INSERT INTO events (id, ring_hash, payload, row_version)
                             VALUES (%s, %s, %s, %s)
                             ON CONFLICT (id) DO UPDATE
                               SET payload = EXCLUDED.payload,
                                   row_version = EXCLUDED.row_version
                               WHERE events.row_version < EXCLUDED.row_version""",
                          rows)
        dst.commit()
        last_id = rows[-1][0]
        m["resume_pk"] = last_id
        json.dump(moves, open("moves.json", "w"))       # checkpoint
        time.sleep(max(0.0, len(rows) / MAX_ROWS_PER_SEC
                            - (time.monotonic() - t0)))

Run one copy_range worker per source node at a time (three concurrent jobs total), so no source ever serves two outbound copies. After the bulk pass, re-run the same loop as a delta pass with an added AND row_version > :snapshot_version predicate until it returns near-zero rows — this drains writes that landed during the hours the bulk copy took.

Operational note: The ON CONFLICT ... WHERE events.row_version < EXCLUDED.row_version guard makes every batch idempotent. A crashed worker resumes from resume_pk and may replay its last batch; the guard turns that replay into a no-op instead of clobbering a fresher row applied by the delta pass.

SRE tip: Watch source p99 latency while the first worker runs and treat MAX_ROWS_PER_SEC as a dial, not a constant. If replica lag on any source exceeds your alert threshold, halve the cap — a rebalance that takes two extra nights is cheaper than a peak-hour brownout. The rate math and adaptive-throttle pattern are covered in depth in the parent guide.

Step 3 — Run a dual-read verification window

Before any router changes behavior, prove the copies are correct under real traffic. Deploy a sampling dual-reader in the data-access layer: for keys whose range appears in moves.json, a configurable fraction of reads also queries the future owner and compares fingerprints:

import random

DUAL_READ_SAMPLE = 0.05          # 5% of reads on moved ranges

def read_event(key, ring_v1, conns, metrics, moves_index):
    node_v1 = ring_v1.owner(key)
    row = fetch(conns[node_v1], key)             # v1 owner stays authoritative
    m = moves_index.lookup(hash32(key))
    if m and random.random() < DUAL_READ_SAMPLE:
        shadow = fetch(conns[m["dst"]], key)
        if fingerprint(row) != fingerprint(shadow):
            metrics.incr("ring_rebalance.dual_read_mismatch",
                         tags={"src": m["src"], "dst": m["dst"]})
    return row

Hold this window for at least one full traffic cycle (24 hours for most services). The gate to proceed is absolute: ring_rebalance.dual_read_mismatch must be zero for the entire window. Early in the window a small mismatch rate is normal — it is the delta pass still draining — but it must trend to zero and stay there.

Operational note: Mismatches on rows that were just written are usually replication-of-copy lag, not corruption: the v1 owner took a write the delta pass has not swept yet. Tag the mismatch metric with row age if you can; persistent mismatches on old rows are the signal that something is genuinely wrong with the copy.

SRE tip: Keep the dual-reader behind a feature flag with the sample rate as a runtime knob. You will want to crank it to 100% for the final hour before the flip on small services, and drop it to zero instantly if the shadow reads themselves push db4/db5 latency into alert territory.

Step 4 — Flip the ring version atomically in etcd

Publish the five-node ring as version 2, then switch the single version pointer with a compare-and-swap so a concurrent operator (or a re-run deploy script) cannot double-apply:

# Publish the v2 ring definition (readable before it is active)
etcdctl put /sharding/ring/v2/nodes '["db1","db2","db3","db4","db5"]'
etcdctl put /sharding/ring/v2/vnodes '64'

# Atomic flip: succeeds only if the active version is still 1
etcdctl txn --interactive=false <<'EOF'
value("/sharding/ring/version") = "1"

put /sharding/ring/version "2"

get /sharding/ring/version
EOF

Routers watch /sharding/ring/version; on the change event they load /sharding/ring/v2/*, rebuild the ring in memory, and start routing moved ranges to db4/db5. Each router should publish its loaded version as a gauge — proceed to cleanup only when every instance reports 2. After the flip, run the delta pass from Step 2 once more against each source to sweep writes that were in flight under v1 during the switch.

Operational note: The transaction’s compare clause is the whole point: if the output shows FAILURE followed by the current version, someone or something already moved the pointer, and blindly re-putting "2" could stomp a rollback to "1" in progress. Treat a failed txn as a stop-and-investigate, not a retry.

SRE tip: Rollback is the same mechanism in reverse — a txn comparing against "2" and putting "1" — and it stays safe exactly as long as the source data has not been deleted. Keep the rollback one-liner in the runbook next to the flip, and rehearse it once on staging while the dual-reader is still running.

Step 5 — Delete moved rows from the source shards

After all routers report v2, the post-flip delta sweep returns zero rows, and the grace period has elapsed, remove the moved ranges from db1–db3 in throttled batches:

#!/usr/bin/env bash
# delete_range.sh <source-host> <lo> <hi> — batched, throttled cleanup
set -euo pipefail
HOST="$1"; LO="$2"; HI="$3"
while :; do
  DELETED=$(psql -h "$HOST" -U rebalance -d appdb -qtAc "
    WITH victims AS (
      SELECT ctid FROM events
      WHERE ring_hash > ${LO} AND ring_hash <= ${HI}
      LIMIT 5000)
    DELETE FROM events WHERE ctid IN (SELECT ctid FROM victims)
    RETURNING 1" | wc -l)
  [ "$DELETED" -eq 0 ] && break
  sleep 0.5
done
psql -h "$HOST" -U rebalance -d appdb -c "VACUUM (ANALYZE) events;"

Drive the script from the same moves.json the copy used, one source node at a time. The 0.5 s sleep caps deletion at ~10 000 rows/s per node; loosen it only if latency dashboards stay green.

Operational note: Deleting 40% of a table leaves substantial dead space; the closing VACUUM (ANALYZE) makes it reusable and refreshes planner statistics, but it does not shrink the file. If you need the disk back immediately, follow up with pg_repack during a quiet window rather than VACUUM FULL, which takes an exclusive lock.

DBA tip: Before the first DELETE, snapshot per-range row counts on both sides one last time and store them with the job record. If anyone questions data integrity weeks later, “source had 41 208 337, destination had 41 208 337, then we deleted” is an answer; “the script finished” is not.

Verification

Run the count-and-checksum pair per range on source and destination before Step 5, using the same predicate boundaries from moves.json:

SELECT COUNT(*)                                             AS row_count,
       SUM(hashtext(id::text || ':' || row_version::text))  AS content_sum
FROM events
WHERE ring_hash > 3221225472 AND ring_hash <= 3251200000;

Expected output — the two nodes must agree exactly:

-- on db2 (source)                     -- on db4 (destination)
 row_count | content_sum                row_count | content_sum
-----------+--------------              -----------+--------------
    412083 | -1830224117                    412083 | -1830224117

Then confirm the fleet is on v2 and the ring is serving from the new nodes:

etcdctl get /sharding/ring/version           # expect: 2
curl -s http://router-1:9100/metrics | grep ring_version
# expect: app_ring_version 2  (on every router instance)

Finally, check that db4 and db5 show ~20% of read QPS each on the traffic dashboard, and that ring_rebalance.dual_read_mismatch stayed at zero through the flip. Any range failing the checksum comparison gets re-run through the Step 2 delta pass and re-verified — never patched by hand.

Failure mode table

Failure mode Root cause SRE mitigation
Move set percentage wildly off (e.g. 78% instead of ~40%) Planner’s hash function or vnode label format differs from the router’s, so computed ownership never matched production reality Assert plan percentage ≈ new/total nodes before copying; extract ring construction into one shared library imported by both router and planner
Dual-read mismatches never reach zero Delta pass predicate uses updated_at with clock skew between nodes, or a writer bypasses row_version bumps, so fresh source writes are never swept Use a monotonic row_version sequence instead of wall-clock time; overlap delta passes by a generous margin and alert if swept-row count fails to decrease pass over pass
Keys intermittently missing after the flip One router instance failed to receive the etcd watch event and still routes moved ranges to old owners, which were then cleaned up Gate Step 5 on every router’s exported ring_version gauge equalling 2; make routers refuse traffic when their watch lease lapses; keep the 24–72 h grace period so rollback needs no restore

FAQ

Should I add both new nodes to the ring at once or one at a time?

Compute the plan for both at once, but execute the copy jobs in whatever order suits your I/O budget. Planning both additions together produces one target ring and one move set, so no range moves twice. Adding db4, rebalancing fully, then adding db5 moves some ranges onto db4 that the second addition immediately moves again to db5 — wasted transfer and a second verification cycle. The only reason to stage additions is if a single new node already saturates your copy-throughput window.

What happens to writes that are in flight during the ring version flip?

A write routed under v1 can land on the old owner milliseconds after the flip. That is why Step 4 ends with a catch-up pass: the delta query re-runs against each source shard until it returns zero rows, sweeping stragglers to the new owner before anything is deleted. Routers watching the etcd key converge within their watch latency — typically well under a second — so the straggler window is tiny, but it must be drained, never assumed empty.

How many virtual nodes per physical node should the ring use?

Between 64 and 256 vnodes per physical node is the practical range. Fewer than 64 gives lumpy ownership, with individual nodes owning 20–30% more keyspace than the mean. More than 512 fragments the move set into thousands of tiny ranges, which bloats job tracking and makes range deletes less efficient. This walkthrough’s 64 vnodes across five nodes keeps per-node ownership within a few percent of even; the tradeoffs are explored further in the consistent hash routing implementation guide.

Back to Shard Rebalancing Data Movement