test(dist): replace Set with DebugInject in rebalance test helpers

Switch rebalance integration tests from using node.Set (quorum write)
to node.DebugInject (test bypass) when seeding keys before triggering
topology changes. This eliminates a ~1-in-50 flake under -shuffle in
CI caused by fan-out transport calls missing deadlines on brand-new
two-node clusters.

Add populateKeysOnAll helper to inject keys across multiple nodes
simultaneously, and migrate leave, replica-diff, and throttle tests
to use the new helper instead of inline key-seeding loops.

Author: hyp3rd

tests/integration/dist_rebalance_leave_test.go

@@ -6,7 +6,6 @@ import (
 	"time"
 
 	"github.com/hyp3rd/hypercache/pkg/backend"
-	cache "github.com/hyp3rd/hypercache/pkg/cache/v2"
 )
 
 // TestDistRebalanceLeave verifies keys are redistributed after a node leaves.
@@ -32,18 +31,15 @@ func TestDistRebalanceLeave(t *testing.T) {
 	nodeC := mustDistNode(ctx, t, "C", addrC, []string{addrA, addrB}, opts...)
 	defer func() { _ = nodeA.Stop(ctx); _ = nodeB.Stop(ctx); _ = nodeC.Stop(ctx) }()
 
-	// Insert keys through A.
-	totalKeys := 300
-	for i := range totalKeys {
-		k := cacheKey(i)
+	// Inject keys across all three nodes — when C leaves, keys
+	// that lived on C need to migrate to surviving owners, so
+	// the pre-leave state must have keys on every node for the
+	// post-leave migration assertion to be meaningful. See
+	// populateKeys' doc in dist_rebalance_test.go for why we use
+	// DebugInject instead of Set.
+	const totalKeys = 300
 
-		it := &cache.Item{Key: k, Value: []byte("v"), Version: 1, Origin: "A", LastUpdated: time.Now()}
-
-		err := nodeA.Set(ctx, it)
-		if err != nil {
-			t.Fatalf("set %s: %v", k, err)
-		}
-	}
+	populateKeysOnAll(ctx, t, totalKeys, nodeA, nodeB, nodeC)
 
 	time.Sleep(250 * time.Millisecond) // allow replication
 

tests/integration/dist_rebalance_replica_diff_test.go

@@ -6,7 +6,6 @@ import (
 	"time"
 
 	"github.com/hyp3rd/hypercache/pkg/backend"
-	cache "github.com/hyp3rd/hypercache/pkg/cache/v2"
 )
 
 // TestDistRebalanceReplicaDiff ensures that when a new replica is added (primary unchanged)
@@ -31,18 +30,14 @@ func TestDistRebalanceReplicaDiff(t *testing.T) {
 	nodeB := mustDistNode(ctx, t, "B", addrB, []string{addrA}, baseOpts...)
 	defer func() { _ = nodeA.Stop(ctx); _ = nodeB.Stop(ctx) }()
 
-	// Insert a set of keys through primary (either node). We'll use A.
-	totalKeys := 200
-	for i := range totalKeys {
-		k := cacheKey(i)
-
-		it := &cache.Item{Key: k, Value: []byte("v"), Version: 1, Origin: "A", LastUpdated: time.Now()}
+	// Inject keys on BOTH A and B (the replica-diff path assumes
+	// replication has already drained across the pre-join cluster
+	// — without that, the post-join replica fan-out has nothing to
+	// diff). See populateKeys' doc in dist_rebalance_test.go for
+	// why we use DebugInject instead of Set.
+	const totalKeys = 200
 
-		err := nodeA.Set(ctx, it)
-		if err != nil {
-			t.Fatalf("set %s: %v", k, err)
-		}
-	}
+	populateKeysOnAll(ctx, t, totalKeys, nodeA, nodeB)
 
 	time.Sleep(300 * time.Millisecond) // allow initial replication
 

tests/integration/dist_rebalance_replica_diff_throttle_test.go

@@ -6,7 +6,6 @@ import (
 	"time"
 
 	"github.com/hyp3rd/hypercache/pkg/backend"
-	cache "github.com/hyp3rd/hypercache/pkg/cache/v2"
 )
 
 // TestDistRebalanceReplicaDiffThrottle ensures the per-tick limit increments throttle metric.
@@ -31,12 +30,12 @@ func TestDistRebalanceReplicaDiffThrottle(t *testing.T) {
 	nodeB := mustDistNode(ctx, t, "B", addrB, []string{addrA}, base...)
 	defer func() { _ = nodeA.Stop(ctx); _ = nodeB.Stop(ctx) }()
 
-	// Seed multiple keys.
-	for i := range 25 {
-		k := cacheKey(i)
-
-		_ = nodeA.Set(ctx, &cache.Item{Key: k, Value: []byte("x"), Version: 1, Origin: "A", LastUpdated: time.Now()})
-	}
+	// Seed keys on BOTH A and B so the post-join replica-diff has
+	// candidates to push to C (without keys on B, the replica-diff
+	// throttle would never fire because only A would have work).
+	// DebugInject also fixes the prior shape's silently-swallowed
+	// Set error.
+	populateKeysOnAll(ctx, t, 25, nodeA, nodeB)
 
 	time.Sleep(250 * time.Millisecond)
 

tests/integration/dist_rebalance_test.go

@@ -30,19 +30,51 @@ func rebalanceTestOpts() []backend.DistMemoryOption {
 	}
 }
 
-// populateKeys writes n test keys to node — used to seed the cluster before
-// triggering a rebalance.
-func populateKeys(ctx context.Context, t *testing.T, node *backend.DistMemory, n int) {
+// populateKeys seeds n test keys onto node's local shard via the
+// DebugInject test bypass — no replication, no quorum check. We use
+// the bypass deliberately: every rebalance test asserts post-state
+// of the ring + migration metrics, not the pre-rebalance
+// replication path. Going through Set would force a quorum write
+// for every key against the brand-new two-node cluster, which is
+// flake-prone (~1 in 50 runs under -shuffle in CI) when a single
+// fan-out transport call misses its deadline.
+//
+// Tests that specifically want to exercise the replication path
+// during populate should call Set directly with appropriate
+// assertions; this helper is for "shape the ring, then test what
+// happens next." Use populateKeysOnAll when the test needs the
+// keys present on multiple nodes (e.g. replica-diff scenarios that
+// assume replication has already drained).
+func populateKeys(_ context.Context, t *testing.T, node *backend.DistMemory, n int) {
 	t.Helper()
 
 	for i := range n {
 		k := cacheKey(i)
+		it := &cache.Item{Key: k, Value: []byte("v"), Version: 1, Origin: "A", LastUpdated: time.Now()}
+
+		node.DebugInject(it)
+	}
+}
 
+// populateKeysOnAll injects n keys on every supplied node via the
+// DebugInject bypass — simulates "replication has already drained
+// across these nodes" without going through the quorum-write path
+// that flakes under -shuffle. Use this for replica-diff and
+// leave-migration tests where the post-topology assertion requires
+// keys to be present on multiple nodes pre-change.
+//
+// Over-replication (a key landing on a node that isn't its actual
+// owner per the ring) is harmless: the rebalance loop sheds keys
+// from nodes that aren't owners after each tick.
+func populateKeysOnAll(_ context.Context, t *testing.T, n int, nodes ...*backend.DistMemory) {
+	t.Helper()
+
+	for i := range n {
+		k := cacheKey(i)
 		it := &cache.Item{Key: k, Value: []byte("v"), Version: 1, Origin: "A", LastUpdated: time.Now()}
 
-		err := node.Set(ctx, it)
-		if err != nil {
-			t.Fatalf("set %s: %v", k, err)
+		for _, node := range nodes {
+			node.DebugInject(it)
 		}
 	}
 }
@@ -124,17 +156,11 @@ func TestDistRebalanceThrottle(t *testing.T) {
 
 	defer func() { _ = nodeA.Stop(ctx); _ = nodeB.Stop(ctx) }()
 
-	// Populate many keys on A.
-	for i := range 400 {
-		k := cacheKey(i)
-
-		it := &cache.Item{Key: k, Value: []byte("v"), Version: 1, Origin: "A", LastUpdated: time.Now()}
-
-		err := nodeA.Set(ctx, it)
-		if err != nil {
-			t.Fatalf("set %s: %v", k, err)
-		}
-	}
+	// Populate many keys on A via DebugInject (test bypass — see
+	// populateKeys' doc for why). We need keys on A's primary
+	// shards so the post-join rebalance has work to do; the
+	// pre-join replication path is not under test here.
+	populateKeys(ctx, t, nodeA, 400)
 
 	// Add third node to force migrations while concurrency=1, which should queue batches.
 	addrC := allocatePort(t)