hash ring

// Package ringhash maps keys to nodes. This step is the naive baseline: hash(key) mod N.
package ringhash

import "hash/fnv"

// hash64 is FNV-1a (64-bit) followed by murmur3's fmix64 finalizer.
//
// Raw FNV-1a is a trap for rings: strings that differ only in their last byte
// ("node-0", "node-1", ...) land within ~2^40 of each other in a 2^64 space, so every
// node clusters on one tiny arc. Modulo hides this because it only looks at the low
// bits; a ring uses the high bits. The finalizer gives full avalanche: flipping one
// input bit flips ~half the output bits. Production rings use murmur3/xxhash directly.
func hash64(s string) uint64 {
	h := fnv.New64a()
	h.Write([]byte(s))
	return fmix64(h.Sum64())
}

func fmix64(k uint64) uint64 {
	k ^= k >> 33
	k *= 0xff51afd7ed558ccd
	k ^= k >> 33
	k *= 0xc4ceb9fe1a85ec53
	k ^= k >> 33
	return k
}

package ringhash

// Modulo assigns a key to nodes[hash(key) % len(nodes)].
// Load is even, but the mapping depends on len(nodes): change N and almost every key moves.
type Modulo struct {
	nodes []string
}

func NewModulo(nodes []string) *Modulo {
	return &Modulo{nodes: append([]string(nil), nodes...)}
}

// Get returns the node that owns key, or "" if there are no nodes.
func (m *Modulo) Get(key string) string {
	if len(m.nodes) == 0 {
		return ""
	}
	return m.nodes[hash64(key)%uint64(len(m.nodes))]
}

package ringhash

import (
	"math/bits"
	"testing"
)

// A ring places nodes by the *high* bits of the hash, so near-identical names must still
// land far apart. Bucket 1,000 sequential names into 16 equal slices of the 64-bit space:
// each slice should get roughly 1000/16 ≈ 62.
func TestHashSpreadsSimilarInputsAcrossTheSpace(t *testing.T) {
	var buckets [16]int
	for _, n := range nodeNames(1_000) {
		buckets[hash64(n)>>60]++
	}
	for i, c := range buckets {
		if c < 35 || c > 95 {
			t.Fatalf("slice %d got %d of 1000 names; hash is clustering (%v)", i, c, buckets)
		}
	}
}

// Avalanche: flipping the last character should flip about 32 of 64 output bits.
func TestHashAvalanche(t *testing.T) {
	total := 0
	names := nodeNames(500)
	for i := 1; i < len(names); i++ {
		total += bits.OnesCount64(hash64(names[i]) ^ hash64(names[i-1]))
	}
	avg := float64(total) / float64(len(names)-1)
	t.Logf("average bits flipped between neighbouring names: %.1f / 64", avg)
	if avg < 28 || avg > 36 {
		t.Fatalf("poor avalanche: %.1f bits flipped on average, want ~32", avg)
	}
}

package ringhash

import "testing"

func TestModuloIsDeterministic(t *testing.T) {
	m := NewModulo(nodeNames(4))
	for _, k := range testKeys(100) {
		if m.Get(k) != m.Get(k) {
			t.Fatalf("key %q mapped to two different nodes", k)
		}
	}
}

func TestModuloEmpty(t *testing.T) {
	if got := NewModulo(nil).Get("anything"); got != "" {
		t.Fatalf("empty Modulo returned %q, want \"\"", got)
	}
}

func TestModuloSpreadsLoadEvenly(t *testing.T) {
	nodes := nodeNames(10)
	m := NewModulo(nodes)
	hi, lo := loadSpread(assignments(testKeys(100_000), m.Get), nodes)
	t.Logf("load: max %.3f× mean, min %.3f× mean", hi, lo)
	if hi > 1.05 || lo < 0.95 {
		t.Fatalf("modulo should be near-uniform; got max %.3f min %.3f", hi, lo)
	}
}

// The reason this step exists: growing the cluster by one node remaps almost every key.
// With N → N+1 nodes a key stays put only if hash mod N == hash mod (N+1), which happens
// for about 1/(N+1) of keys — so roughly N/(N+1) of them move.
func TestModuloAddingANodeMovesMostKeys(t *testing.T) {
	keys := testKeys(100_000)
	before := assignments(keys, NewModulo(nodeNames(4)).Get)
	after := assignments(keys, NewModulo(nodeNames(5)).Get)
	moved := movedFraction(before, after)
	t.Logf("4 → 5 nodes: %.1f%% of keys moved (ideal would be 20%%)", moved*100)
	if moved < 0.75 {
		t.Fatalf("expected ~80%% of keys to move, got %.1f%%", moved*100)
	}
}

package ringhash

import (
	"slices"
	"sort"
)

// Ring is a consistent-hash ring with one point per node. Nodes and keys are hashed onto
// the same 64-bit circle; a key belongs to the first node at or after it, clockwise.
//
// Not safe for concurrent use; wrap with a sync.RWMutex if nodes change while serving.
type Ring struct {
	points []uint64          // sorted node positions
	owner  map[uint64]string // position → node
}

func NewRing() *Ring {
	return &Ring{owner: make(map[uint64]string)}
}

// Add places node on the ring. Adding an existing node is a no-op.
func (r *Ring) Add(node string) {
	p := hash64(node)
	if _, ok := r.owner[p]; ok {
		return
	}
	r.owner[p] = node
	i, _ := slices.BinarySearch(r.points, p)
	r.points = slices.Insert(r.points, i, p)
}

// Remove takes node off the ring; its keys fall through to the next node clockwise.
func (r *Ring) Remove(node string) {
	p := hash64(node)
	if r.owner[p] != node {
		return
	}
	delete(r.owner, p)
	i, _ := slices.BinarySearch(r.points, p)
	r.points = slices.Delete(r.points, i, i+1)
}

// Get returns the node that owns key, or "" if the ring is empty. O(log n).
func (r *Ring) Get(key string) string {
	if len(r.points) == 0 {
		return ""
	}
	h := hash64(key)
	i := sort.Search(len(r.points), func(i int) bool { return r.points[i] >= h })
	if i == len(r.points) {
		i = 0 // past the last point: wrap around to the first
	}
	return r.owner[r.points[i]]
}

// Nodes returns the nodes on the ring in ring order.
func (r *Ring) Nodes() []string {
	out := make([]string, len(r.points))
	for i, p := range r.points {
		out[i] = r.owner[p]
	}
	return out
}

package ringhash

import (
	"slices"
	"testing"
)

func newRingWith(nodes []string) *Ring {
	r := NewRing()
	for _, n := range nodes {
		r.Add(n)
	}
	return r
}

func TestRingEmpty(t *testing.T) {
	if got := NewRing().Get("k"); got != "" {
		t.Fatalf("empty ring returned %q, want \"\"", got)
	}
}

func TestRingSingleNodeOwnsEverything(t *testing.T) {
	r := newRingWith([]string{"solo"})
	for _, k := range testKeys(1_000) {
		if got := r.Get(k); got != "solo" {
			t.Fatalf("key %q → %q, want solo", k, got)
		}
	}
}

// A key belongs to the first node clockwise from the key's position, wrapping past the top.
func TestRingOwnerIsFirstNodeClockwise(t *testing.T) {
	r := newRingWith(nodeNames(5))
	positions := make([]uint64, 0, 5)
	owner := map[uint64]string{}
	for _, n := range nodeNames(5) {
		p := hash64(n)
		positions = append(positions, p)
		owner[p] = n
	}
	slices.Sort(positions)
	for _, k := range testKeys(2_000) {
		h := hash64(k)
		want := owner[positions[0]] // wrap-around default
		for _, p := range positions {
			if p >= h {
				want = owner[p]
				break
			}
		}
		if got := r.Get(k); got != want {
			t.Fatalf("key %q (hash %d) → %q, want %q", k, h, got, want)
		}
	}
}

func TestRingAddIsIdempotent(t *testing.T) {
	r := newRingWith([]string{"a", "b"})
	r.Add("a")
	if got := len(r.Nodes()); got != 2 {
		t.Fatalf("adding an existing node changed node count to %d", got)
	}
}

// The headline property: adding a node only moves keys *to the new node*, and only
// from the one neighbour whose arc it splits. Nothing moves between existing nodes.
func TestRingAddingANodeOnlyMovesKeysToIt(t *testing.T) {
	keys := testKeys(100_000)
	r := newRingWith(nodeNames(4))
	before := assignments(keys, r.Get)
	r.Add("node-4")
	after := assignments(keys, r.Get)

	for k, was := range before {
		if now := after[k]; now != was && now != "node-4" {
			t.Fatalf("key %q moved %s → %s; only moves to the new node are allowed", k, was, now)
		}
	}
	t.Logf("4 → 5 nodes: %.1f%% of keys moved (modulo moved ~80%%)", movedFraction(before, after)*100)
}

func TestRingRemovingANodeOnlyMovesItsKeys(t *testing.T) {
	keys := testKeys(100_000)
	r := newRingWith(nodeNames(5))
	before := assignments(keys, r.Get)
	r.Remove("node-2")
	after := assignments(keys, r.Get)

	for k, was := range before {
		if was != "node-2" && after[k] != was {
			t.Fatalf("key %q moved off surviving node %s", k, was)
		}
		if after[k] == "node-2" {
			t.Fatalf("key %q still maps to removed node", k)
		}
	}
}

// The weakness this step leaves behind: with one point per node, arc lengths are random,
// so load is badly skewed. Step 03 fixes it with virtual nodes.
func TestRingLoadIsUneven(t *testing.T) {
	nodes := nodeNames(10)
	hi, lo := loadSpread(assignments(testKeys(100_000), newRingWith(nodes).Get), nodes)
	t.Logf("load: max %.2f× mean, min %.2f× mean", hi, lo)
	if hi < 1.3 {
		t.Fatalf("expected visible skew with one point per node; max was only %.2f× mean", hi)
	}
}

package ringhash

import (
	"slices"
	"sort"
	"strconv"
)

// Ring is a consistent-hash ring with virtual nodes: each physical node is hashed onto
// the circle `vnodes` times (as "node#0", "node#1", ...). Many small arcs per node average
// out the randomness of one big arc, so load evens out, and a departing node's keys
// scatter across many neighbours instead of landing on one.
//
// Not safe for concurrent use; wrap with a sync.RWMutex if nodes change while serving.
type Ring struct {
	vnodes int
	points []uint64          // sorted vnode positions
	owner  map[uint64]string // position → physical node
	nodes  map[string]bool
}

// NewRing returns an empty ring that places `vnodes` points per node (minimum 1).
func NewRing(vnodes int) *Ring {
	return &Ring{vnodes: max(vnodes, 1), owner: make(map[uint64]string), nodes: make(map[string]bool)}
}

func vnodeKey(node string, i int) string { return node + "#" + strconv.Itoa(i) }

// Add places node's vnodes on the ring. Adding an existing node is a no-op.
// O(v log P) for v vnodes and P points, plus a re-sort.
func (r *Ring) Add(node string) {
	if r.nodes[node] {
		return
	}
	r.nodes[node] = true
	for i := range r.vnodes {
		p := hash64(vnodeKey(node, i))
		if _, taken := r.owner[p]; taken {
			continue // 64-bit collision: astronomically rare; first owner keeps the point
		}
		r.owner[p] = node
		r.points = append(r.points, p)
	}
	slices.Sort(r.points)
}

// Remove takes all of node's vnodes off the ring.
func (r *Ring) Remove(node string) {
	if !r.nodes[node] {
		return
	}
	delete(r.nodes, node)
	r.points = slices.DeleteFunc(r.points, func(p uint64) bool {
		if r.owner[p] == node {
			delete(r.owner, p)
			return true
		}
		return false
	})
}

// Get returns the node that owns key, or "" if the ring is empty. O(log P).
func (r *Ring) Get(key string) string {
	if len(r.points) == 0 {
		return ""
	}
	h := hash64(key)
	i := sort.Search(len(r.points), func(i int) bool { return r.points[i] >= h })
	if i == len(r.points) {
		i = 0
	}
	return r.owner[r.points[i]]
}

// Len is the number of points (vnodes) on the ring.
func (r *Ring) Len() int { return len(r.points) }

@@ -3,44 +3,63 @@
 import (
 	"slices"
 	"sort"
+	"strconv"
 )
 
-// Ring is a consistent-hash ring with one point per node. Nodes and keys are hashed onto
-// the same 64-bit circle; a key belongs to the first node at or after it, clockwise.
+// Ring is a consistent-hash ring with virtual nodes: each physical node is hashed onto
+// the circle `vnodes` times (as "node#0", "node#1", ...). Many small arcs per node average
+// out the randomness of one big arc, so load evens out, and a departing node's keys
+// scatter across many neighbours instead of landing on one.
 //
 // Not safe for concurrent use; wrap with a sync.RWMutex if nodes change while serving.
 type Ring struct {
-	points []uint64          // sorted node positions
-	owner  map[uint64]string // position → node
+	vnodes int
+	points []uint64          // sorted vnode positions
+	owner  map[uint64]string // position → physical node
+	nodes  map[string]bool
 }
 
-func NewRing() *Ring {
-	return &Ring{owner: make(map[uint64]string)}
+// NewRing returns an empty ring that places `vnodes` points per node (minimum 1).
+func NewRing(vnodes int) *Ring {
+	return &Ring{vnodes: max(vnodes, 1), owner: make(map[uint64]string), nodes: make(map[string]bool)}
 }
 
-// Add places node on the ring. Adding an existing node is a no-op.
+func vnodeKey(node string, i int) string { return node + "#" + strconv.Itoa(i) }
+
+// Add places node's vnodes on the ring. Adding an existing node is a no-op.
+// O(v log P) for v vnodes and P points, plus a re-sort.
 func (r *Ring) Add(node string) {
-	p := hash64(node)
-	if _, ok := r.owner[p]; ok {
+	if r.nodes[node] {
 		return
 	}
-	r.owner[p] = node
-	i, _ := slices.BinarySearch(r.points, p)
-	r.points = slices.Insert(r.points, i, p)
+	r.nodes[node] = true
+	for i := range r.vnodes {
+		p := hash64(vnodeKey(node, i))
+		if _, taken := r.owner[p]; taken {
+			continue // 64-bit collision: astronomically rare; first owner keeps the point
+		}
+		r.owner[p] = node
+		r.points = append(r.points, p)
+	}
+	slices.Sort(r.points)
 }
 
-// Remove takes node off the ring; its keys fall through to the next node clockwise.
+// Remove takes all of node's vnodes off the ring.
 func (r *Ring) Remove(node string) {
-	p := hash64(node)
-	if r.owner[p] != node {
+	if !r.nodes[node] {
 		return
 	}
-	delete(r.owner, p)
-	i, _ := slices.BinarySearch(r.points, p)
-	r.points = slices.Delete(r.points, i, i+1)
+	delete(r.nodes, node)
+	r.points = slices.DeleteFunc(r.points, func(p uint64) bool {
+		if r.owner[p] == node {
+			delete(r.owner, p)
+			return true
+		}
+		return false
+	})
 }
 
-// Get returns the node that owns key, or "" if the ring is empty. O(log n).
+// Get returns the node that owns key, or "" if the ring is empty. O(log P).
 func (r *Ring) Get(key string) string {
 	if len(r.points) == 0 {
 		return ""
@@ -48,16 +67,10 @@
 	h := hash64(key)
 	i := sort.Search(len(r.points), func(i int) bool { return r.points[i] >= h })
 	if i == len(r.points) {
-		i = 0 // past the last point: wrap around to the first
+		i = 0
 	}
 	return r.owner[r.points[i]]
 }
 
-// Nodes returns the nodes on the ring in ring order.
-func (r *Ring) Nodes() []string {
-	out := make([]string, len(r.points))
-	for i, p := range r.points {
-		out[i] = r.owner[p]
-	}
-	return out
-}
+// Len is the number of points (vnodes) on the ring.
+func (r *Ring) Len() int { return len(r.points) }

package ringhash

import (
	"testing"
)

func newRingWith(vnodes int, nodes []string) *Ring {
	r := NewRing(vnodes)
	for _, n := range nodes {
		r.Add(n)
	}
	return r
}

func TestRingEmpty(t *testing.T) {
	if got := NewRing(100).Get("k"); got != "" {
		t.Fatalf("empty ring returned %q, want \"\"", got)
	}
}

func TestRingSingleNodeOwnsEverything(t *testing.T) {
	r := newRingWith(100, []string{"solo"})
	for _, k := range testKeys(1_000) {
		if got := r.Get(k); got != "solo" {
			t.Fatalf("key %q → %q, want solo", k, got)
		}
	}
}

func TestRingPlacesVnodesPerNode(t *testing.T) {
	r := newRingWith(50, nodeNames(4))
	if got := r.Len(); got != 200 {
		t.Fatalf("4 nodes × 50 vnodes: ring has %d points, want 200", got)
	}
	r.Add("node-0")
	if got := r.Len(); got != 200 {
		t.Fatalf("re-adding a node changed the point count to %d", got)
	}
	r.Remove("node-1")
	if got := r.Len(); got != 150 {
		t.Fatalf("after removing one node: %d points, want 150", got)
	}
}

func TestRingAddingANodeOnlyMovesKeysToIt(t *testing.T) {
	keys := testKeys(100_000)
	r := newRingWith(160, nodeNames(4))
	before := assignments(keys, r.Get)
	r.Add("node-4")
	after := assignments(keys, r.Get)

	for k, was := range before {
		if now := after[k]; now != was && now != "node-4" {
			t.Fatalf("key %q moved %s → %s; only moves to the new node are allowed", k, was, now)
		}
	}
	moved := movedFraction(before, after)
	t.Logf("4 → 5 nodes: %.1f%% of keys moved (ideal 1/5 = 20%%)", moved*100)
	if moved < 0.15 || moved > 0.25 {
		t.Fatalf("with vnodes the new node should take ~20%% of keys, took %.1f%%", moved*100)
	}
}

// With one point per node, a removed node's keys all land on its single clockwise
// neighbour. With vnodes they scatter across the cluster, so no one node absorbs the hit.
func TestRingRemovalSpreadsKeysAcrossSurvivors(t *testing.T) {
	keys := testKeys(100_000)
	r := newRingWith(160, nodeNames(5))
	before := assignments(keys, r.Get)
	r.Remove("node-2")
	after := assignments(keys, r.Get)

	gainedBy := map[string]int{}
	for k, was := range before {
		if was != "node-2" && after[k] != was {
			t.Fatalf("key %q moved off surviving node %s", k, was)
		}
		if was == "node-2" {
			gainedBy[after[k]]++
		}
	}
	t.Logf("node-2's keys went to: %v", gainedBy)
	if len(gainedBy) != 4 {
		t.Fatalf("removed node's keys should spread over all 4 survivors, went to %d", len(gainedBy))
	}
}

func TestRingVnodesEvenOutLoad(t *testing.T) {
	nodes := nodeNames(10)
	hi, lo := loadSpread(assignments(testKeys(100_000), newRingWith(160, nodes).Get), nodes)
	t.Logf("160 vnodes/node: max %.2f× mean, min %.2f× mean", hi, lo)
	if hi > 1.25 || lo < 0.75 {
		t.Fatalf("160 vnodes should keep every node within ±25%% of mean; got %.2f / %.2f", hi, lo)
	}
}

// Not an assertion — prints the trade-off curve the lesson quotes: more vnodes, tighter
// balance, more memory and slower membership changes.
func TestRingVnodeSweep(t *testing.T) {
	nodes := nodeNames(10)
	keys := testKeys(100_000)
	for _, v := range []int{1, 10, 50, 100, 200, 500} {
		hi, lo := loadSpread(assignments(keys, newRingWith(v, nodes).Get), nodes)
		t.Logf("vnodes=%-4d points=%-5d max %.2f× min %.2f×", v, v*len(nodes), hi, lo)
	}
}

func BenchmarkRingGet(b *testing.B) {
	r := newRingWith(160, nodeNames(10))
	keys := testKeys(1_024)
	b.ResetTimer()
	for i := 0; b.Loop(); i++ {
		r.Get(keys[i%len(keys)])
	}
}

@@ -1,12 +1,11 @@
 package ringhash
 
 import (
-	"slices"
 	"testing"
 )
 
-func newRingWith(nodes []string) *Ring {
-	r := NewRing()
+func newRingWith(vnodes int, nodes []string) *Ring {
+	r := NewRing(vnodes)
 	for _, n := range nodes {
 		r.Add(n)
 	}
@@ -14,13 +13,13 @@
 }
 
 func TestRingEmpty(t *testing.T) {
-	if got := NewRing().Get("k"); got != "" {
+	if got := NewRing(100).Get("k"); got != "" {
 		t.Fatalf("empty ring returned %q, want \"\"", got)
 	}
 }
 
 func TestRingSingleNodeOwnsEverything(t *testing.T) {
-	r := newRingWith([]string{"solo"})
+	r := newRingWith(100, []string{"solo"})
 	for _, k := range testKeys(1_000) {
 		if got := r.Get(k); got != "solo" {
 			t.Fatalf("key %q → %q, want solo", k, got)
@@ -28,45 +27,24 @@
 	}
 }
 
-// A key belongs to the first node clockwise from the key's position, wrapping past the top.
-func TestRingOwnerIsFirstNodeClockwise(t *testing.T) {
-	r := newRingWith(nodeNames(5))
-	positions := make([]uint64, 0, 5)
-	owner := map[uint64]string{}
-	for _, n := range nodeNames(5) {
-		p := hash64(n)
-		positions = append(positions, p)
-		owner[p] = n
+func TestRingPlacesVnodesPerNode(t *testing.T) {
+	r := newRingWith(50, nodeNames(4))
+	if got := r.Len(); got != 200 {
+		t.Fatalf("4 nodes × 50 vnodes: ring has %d points, want 200", got)
 	}
-	slices.Sort(positions)
-	for _, k := range testKeys(2_000) {
-		h := hash64(k)
-		want := owner[positions[0]] // wrap-around default
-		for _, p := range positions {
-			if p >= h {
-				want = owner[p]
-				break
-			}
-		}
-		if got := r.Get(k); got != want {
-			t.Fatalf("key %q (hash %d) → %q, want %q", k, h, got, want)
-		}
+	r.Add("node-0")
+	if got := r.Len(); got != 200 {
+		t.Fatalf("re-adding a node changed the point count to %d", got)
 	}
-}
-
-func TestRingAddIsIdempotent(t *testing.T) {
-	r := newRingWith([]string{"a", "b"})
-	r.Add("a")
-	if got := len(r.Nodes()); got != 2 {
-		t.Fatalf("adding an existing node changed node count to %d", got)
+	r.Remove("node-1")
+	if got := r.Len(); got != 150 {
+		t.Fatalf("after removing one node: %d points, want 150", got)
 	}
 }
 
-// The headline property: adding a node only moves keys *to the new node*, and only
-// from the one neighbour whose arc it splits. Nothing moves between existing nodes.
 func TestRingAddingANodeOnlyMovesKeysToIt(t *testing.T) {
 	keys := testKeys(100_000)
-	r := newRingWith(nodeNames(4))
+	r := newRingWith(160, nodeNames(4))
 	before := assignments(keys, r.Get)
 	r.Add("node-4")
 	after := assignments(keys, r.Get)
@@ -76,33 +54,62 @@
 			t.Fatalf("key %q moved %s → %s; only moves to the new node are allowed", k, was, now)
 		}
 	}
-	t.Logf("4 → 5 nodes: %.1f%% of keys moved (modulo moved ~80%%)", movedFraction(before, after)*100)
+	moved := movedFraction(before, after)
+	t.Logf("4 → 5 nodes: %.1f%% of keys moved (ideal 1/5 = 20%%)", moved*100)
+	if moved < 0.15 || moved > 0.25 {
+		t.Fatalf("with vnodes the new node should take ~20%% of keys, took %.1f%%", moved*100)
+	}
 }
 
-func TestRingRemovingANodeOnlyMovesItsKeys(t *testing.T) {
+// With one point per node, a removed node's keys all land on its single clockwise
+// neighbour. With vnodes they scatter across the cluster, so no one node absorbs the hit.
+func TestRingRemovalSpreadsKeysAcrossSurvivors(t *testing.T) {
 	keys := testKeys(100_000)
-	r := newRingWith(nodeNames(5))
+	r := newRingWith(160, nodeNames(5))
 	before := assignments(keys, r.Get)
 	r.Remove("node-2")
 	after := assignments(keys, r.Get)
 
+	gainedBy := map[string]int{}
 	for k, was := range before {
 		if was != "node-2" && after[k] != was {
 			t.Fatalf("key %q moved off surviving node %s", k, was)
 		}
-		if after[k] == "node-2" {
-			t.Fatalf("key %q still maps to removed node", k)
+		if was == "node-2" {
+			gainedBy[after[k]]++
 		}
 	}
+	t.Logf("node-2's keys went to: %v", gainedBy)
+	if len(gainedBy) != 4 {
+		t.Fatalf("removed node's keys should spread over all 4 survivors, went to %d", len(gainedBy))
+	}
 }
 
-// The weakness this step leaves behind: with one point per node, arc lengths are random,
-// so load is badly skewed. Step 03 fixes it with virtual nodes.
-func TestRingLoadIsUneven(t *testing.T) {
+func TestRingVnodesEvenOutLoad(t *testing.T) {
 	nodes := nodeNames(10)
-	hi, lo := loadSpread(assignments(testKeys(100_000), newRingWith(nodes).Get), nodes)
-	t.Logf("load: max %.2f× mean, min %.2f× mean", hi, lo)
-	if hi < 1.3 {
-		t.Fatalf("expected visible skew with one point per node; max was only %.2f× mean", hi)
+	hi, lo := loadSpread(assignments(testKeys(100_000), newRingWith(160, nodes).Get), nodes)
+	t.Logf("160 vnodes/node: max %.2f× mean, min %.2f× mean", hi, lo)
+	if hi > 1.25 || lo < 0.75 {
+		t.Fatalf("160 vnodes should keep every node within ±25%% of mean; got %.2f / %.2f", hi, lo)
 	}
 }
+
+// Not an assertion — prints the trade-off curve the lesson quotes: more vnodes, tighter
+// balance, more memory and slower membership changes.
+func TestRingVnodeSweep(t *testing.T) {
+	nodes := nodeNames(10)
+	keys := testKeys(100_000)
+	for _, v := range []int{1, 10, 50, 100, 200, 500} {
+		hi, lo := loadSpread(assignments(keys, newRingWith(v, nodes).Get), nodes)
+		t.Logf("vnodes=%-4d points=%-5d max %.2f× min %.2f×", v, v*len(nodes), hi, lo)
+	}
+}
+
+func BenchmarkRingGet(b *testing.B) {
+	r := newRingWith(160, nodeNames(10))
+	keys := testKeys(1_024)
+	b.ResetTimer()
+	for i := 0; b.Loop(); i++ {
+		r.Get(keys[i%len(keys)])
+	}
+}