test: finish test-suite cleanup WP-3..WP-12 and promote shipped designs

adapter/admin_backup_renew_race_test.go

@@ -72,38 +72,6 @@ func TestRenewBackupKeepsAConcurrentlyRenewedSession(t *testing.T) {
 	require.NotContains(t, proposer.subtypes(), backupSubtypeUnreserve)
 }
 
-// With no concurrent renewal the failing attempt still owns the session, so it
-// must tear the pin down exactly as before.
-func TestRenewBackupStillReleasesWhenItOwnsTheSession(t *testing.T) {
-	t.Parallel()
-
-	group := &backupTestGroup{status: raftengine.Status{AppliedIndex: 100}, every: 10_000}
-	proposer := newBackupTestProposer()
-	srv := newBackupControlTestServer(
-		t,
-		&backupTestStore{},
-		map[uint64]*backupTestGroup{1: group},
-		map[uint64]*backupTestProposer{1: proposer},
-		nil,
-	)
-	begin, err := srv.BeginBackup(context.Background(), &pb.BeginBackupRequest{})
-	require.NoError(t, err)
-	tok, err := srv.decodeBackupToken(begin.GetPinToken())
-	require.NoError(t, err)
-
-	proposer.mu.Lock()
-	proposer.failures[backupSubtypePin] = 8
-	proposer.transportError[backupSubtypePin] = stderrors.New("leader unavailable")
-	proposer.mu.Unlock()
-
-	_, err = srv.RenewBackup(context.Background(), &pb.RenewBackupRequest{PinToken: begin.GetPinToken()})
-	require.Equal(t, codes.Unavailable, status.Code(err))
-	require.Contains(t, proposer.subtypes(), backupSubtypeRelease)
-	require.Contains(t, proposer.subtypes(), backupSubtypeUnreserve)
-	_, err = srv.backupRouteSnapshotForToken(tok)
-	require.Equal(t, codes.FailedPrecondition, status.Code(err))
-}
-
 // A reservation whose proposal fails for any reason other than a capacity
 // rejection is ambiguous: it may have committed with only the response lost.
 // Leaving it in place holds one of the few global active-backup slots until

adapter/admin_backup_test.go

@@ -1210,17 +1210,6 @@ func TestRenewBackupReleasesResourcesAfterPinFanoutFailure(t *testing.T) {
 	require.Equal(t, codes.FailedPrecondition, status.Code(err))
 }
 
-func TestBeginBackupMapsCapacityReservationToResourceExhausted(t *testing.T) {
-	t.Parallel()
-	group := &backupTestGroup{status: raftengine.Status{AppliedIndex: 100}, every: 10_000}
-	proposer := newBackupTestProposer()
-	proposer.responseError[backupSubtypeReserve] = kv.ErrTooManyActiveBackups
-	srv := newBackupControlTestServer(t, &backupTestStore{}, map[uint64]*backupTestGroup{1: group}, map[uint64]*backupTestProposer{1: proposer}, nil)
-
-	_, err := srv.BeginBackup(context.Background(), &pb.BeginBackupRequest{})
-	require.Equal(t, codes.ResourceExhausted, status.Code(err))
-}
-
 func TestBeginBackupMapsForwardedCapacityReservationStatusToResourceExhausted(t *testing.T) {
 	t.Parallel()
 	group := &backupTestGroup{status: raftengine.Status{AppliedIndex: 100}, every: 10_000}

adapter/admin_grpc_test.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/bootjp/elastickv/internal/raftengine"
+	"github.com/bootjp/elastickv/internal/testutil"
 	pb "github.com/bootjp/elastickv/proto"
 	"github.com/stretchr/testify/require"
 	"google.golang.org/grpc"
@@ -207,11 +208,7 @@ func TestGetRaftGroupsLeaderVersionAsync(t *testing.T) {
 		t.Fatalf("GetRaftGroups first: %v", err)
 	}
 	requireLeaderVersion(t, resp, "", "first")
-	select {
-	case <-probeStarted:
-	case <-time.After(time.Second):
-		t.Fatal("leader version probe was not started")
-	}
+	testutil.Recv(t, probeStarted, time.Second, "leader version probe was not started")
 	second, err := srv.GetRaftGroups(context.Background(), &pb.GetRaftGroupsRequest{})
 	if err != nil {
 		t.Fatalf("GetRaftGroups second: %v", err)
@@ -498,11 +495,7 @@ func waitForLeaderVersion(t *testing.T, srv *AdminServer, want string) {
 		if len(resp.Groups) == 1 && resp.Groups[0].LeaderNodeVersion == want {
 			return
 		}
-		select {
-		case <-deadline:
-			t.Fatalf("leader version cache was not populated with %q", want)
-		case <-time.After(10 * time.Millisecond):
-		}
+		testutil.NoRecv(t, deadline, 10*time.Millisecond, "leader version cache was not populated with %q", want)
 	}
 }
 
@@ -528,13 +521,9 @@ func TestGetRaftGroupsLeaderVersionProbePropagatesAuthMetadata(t *testing.T) {
 	if _, err := srv.GetRaftGroups(ctx, &pb.GetRaftGroupsRequest{}); err != nil {
 		t.Fatalf("GetRaftGroups: %v", err)
 	}
-	select {
-	case got := <-gotAuth:
-		if len(got) != 1 || got[0] != "Bearer admin-token" {
-			t.Fatalf("authorization metadata = %v, want bearer token", got)
-		}
-	case <-time.After(time.Second):
-		t.Fatal("leader version probe did not run")
+	got := testutil.Recv(t, gotAuth, time.Second, "leader version probe did not run")
+	if len(got) != 1 || got[0] != "Bearer admin-token" {
+		t.Fatalf("authorization metadata = %v, want bearer token", got)
 	}
 }
 
@@ -746,11 +735,7 @@ func TestFanoutConfigurationCallsBoundedConcurrency(t *testing.T) {
 	}
 
 	close(release) // let everything drain
-	select {
-	case <-done:
-	case <-time.After(5 * time.Second):
-		t.Fatal("fanoutConfigurationCalls did not return after release")
-	}
+	testutil.Closed(t, done, 5*time.Second, "fanoutConfigurationCalls did not return after release")
 }
 
 // testProbeGroup is a minimal AdminGroup for concurrency tests; it runs cb
@@ -1137,13 +1122,9 @@ func TestCollectLiveMembersHonoursCtxCancel(t *testing.T) {
 		resCh <- collectLiveMembers(ctx, groups, "self")
 	}()
 
-	select {
-	case r := <-resCh:
-		// With ctx already cancelled, no live config landed; expect empty maps.
-		if len(r.addrByID) != 0 || len(r.order) != 0 {
-			t.Fatalf("expected empty results on early cancel, got addrByID=%v order=%v", r.addrByID, r.order)
-		}
-	case <-time.After(1 * time.Second):
-		t.Fatal("collectLiveMembers blocked past 1s despite cancelled ctx — wg.Wait() regression?")
+	r := testutil.Recv(t, resCh, 1*time.Second, "collectLiveMembers blocked past 1s despite cancelled ctx — wg.Wait() regression?")
+	// With ctx already cancelled, no live config landed; expect empty maps.
+	if len(r.addrByID) != 0 || len(r.order) != 0 {
+		t.Fatalf("expected empty results on early cancel, got addrByID=%v order=%v", r.addrByID, r.order)
 	}
 }

adapter/background_reservation_shutdown_test.go

@@ -5,6 +5,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/bootjp/elastickv/internal/testutil"
 	"github.com/stretchr/testify/require"
 )
 
@@ -18,11 +19,7 @@ func TestBackgroundReservationStartsLiveAfterShutdownBound(t *testing.T) {
 	ran := make(chan error, 1)
 	r.start(time.Second, func(ctx context.Context) { ran <- ctx.Err() })
 	require.NoError(t, <-ran)
-	select {
-	case <-done:
-	case <-time.After(time.Second):
-		t.Fatal("stop did not join reserved cleanup")
-	}
+	testutil.Closed(t, done, time.Second, "stop did not join reserved cleanup")
 }
 
 func TestBackgroundReservationCannotStartAfterStopReturns(t *testing.T) {
@@ -34,9 +31,5 @@ func TestBackgroundReservationCannotStartAfterStopReturns(t *testing.T) {
 	r.start(time.Second, func(context.Context) { ran <- struct{}{} })
 	r.release()
 	require.Zero(t, w.gate.remaining())
-	select {
-	case <-ran:
-		t.Fatal("cleanup started after shutdown returned")
-	case <-time.After(time.Millisecond):
-	}
+	testutil.NoRecv(t, ran, time.Millisecond, "cleanup started after shutdown returned")
 }

adapter/distribution_server_test.go

@@ -14,7 +14,9 @@ import (
 	"github.com/bootjp/elastickv/distribution"
 	"github.com/bootjp/elastickv/internal/fskeys"
 	"github.com/bootjp/elastickv/internal/raftengine"
+	"github.com/bootjp/elastickv/internal/raftengine/raftenginetest"
 	"github.com/bootjp/elastickv/internal/s3keys"
+	"github.com/bootjp/elastickv/internal/testutil"
 	"github.com/bootjp/elastickv/kv"
 	pb "github.com/bootjp/elastickv/proto"
 	"github.com/bootjp/elastickv/store"
@@ -317,7 +319,7 @@ func TestDistributionServerPendingLegacyResolutionMode(t *testing.T) {
 func TestDistributionServerPendingLegacyResolutionRoutedWrappers(t *testing.T) {
 	alloc := &pendingResolutionAllocator{distributionTSOAllocator: &distributionTSOAllocator{leader: true}, pendingFloor: 100}
 	addr := serveDistributionTestServer(t, NewDistributionServer(distribution.NewEngine(), nil, WithDistributionTimestampAllocator(alloc)))
-	routed, err := kv.NewLeaderRoutedTSOAllocator(&distributionTSOAllocator{}, distributionLeaderView{addr: addr})
+	routed, err := kv.NewLeaderRoutedTSOAllocator(&distributionTSOAllocator{}, distributionLeaderView(addr))
 	require.NoError(t, err)
 	t.Cleanup(func() { require.NoError(t, routed.Close()) })
 	batch, err := kv.NewBatchAllocator(routed, 2)
@@ -351,7 +353,7 @@ func TestDistributionServerValidateTimestamp_LeaderRoutedRPC(t *testing.T) {
 	))
 	routed, err := kv.NewLeaderRoutedTSOAllocator(
 		&distributionTSOAllocator{leader: false},
-		distributionLeaderView{addr: addr},
+		distributionLeaderView(addr),
 	)
 	require.NoError(t, err)
 	t.Cleanup(func() { require.NoError(t, routed.Close()) })
@@ -409,7 +411,7 @@ func TestDistributionServerGetTimestamp_LeaderRoutedRPC(t *testing.T) {
 	))
 
 	local := &distributionTSOAllocator{leader: false}
-	routed, err := kv.NewLeaderRoutedTSOAllocator(local, distributionLeaderView{addr: addr})
+	routed, err := kv.NewLeaderRoutedTSOAllocator(local, distributionLeaderView(addr))
 	require.NoError(t, err)
 	t.Cleanup(func() { require.NoError(t, routed.Close()) })
 
@@ -423,7 +425,7 @@ func TestDistributionServerGetTimestamp_LeaderRoutedRPC(t *testing.T) {
 func TestDistributionServerGetTimestamp_LeaderRoutedRejectsLegacyServer(t *testing.T) {
 	addr := serveDistributionTestServer(t, NewDistributionServer(distribution.NewEngine(), nil))
 	local := &distributionTSOAllocator{leader: false}
-	routed, err := kv.NewLeaderRoutedTSOAllocator(local, distributionLeaderView{addr: addr})
+	routed, err := kv.NewLeaderRoutedTSOAllocator(local, distributionLeaderView(addr))
 	require.NoError(t, err)
 	t.Cleanup(func() { require.NoError(t, routed.Close()) })
 
@@ -466,7 +468,7 @@ func TestDistributionServerGetTimestamp_LeaderRoutedActivatesCutover(t *testing.
 	local := &distributionTSOAllocator{leader: false}
 	routed, err := kv.NewLeaderRoutedTSOAllocator(
 		local,
-		distributionLeaderView{addr: addr},
+		distributionLeaderView(addr),
 		kv.WithTSOCutoverActivation(),
 	)
 	require.NoError(t, err)
@@ -507,7 +509,7 @@ func (legacyIgnoringTimestampValidationServer) ValidateTimestamp(context.Context
 
 func TestDistributionServerPendingLegacyResolutionRejectsOldModeIgnoringServer(t *testing.T) {
 	addr := serveDistributionTestServer(t, legacyIgnoringTimestampValidationServer{})
-	routed, err := kv.NewLeaderRoutedTSOAllocator(&distributionTSOAllocator{}, distributionLeaderView{addr: addr})
+	routed, err := kv.NewLeaderRoutedTSOAllocator(&distributionTSOAllocator{}, distributionLeaderView(addr))
 	require.NoError(t, err)
 	t.Cleanup(func() { require.NoError(t, routed.Close()) })
 	require.ErrorIs(t, routed.ValidateLegacyResolutionTimestamps(context.Background(), 50, 100), kv.ErrTSOProtocolUnsupported)
@@ -521,18 +523,6 @@ func TestNewDistributionServer_DefaultCatalogReloadRetryPolicy(t *testing.T) {
 	require.Equal(t, defaultCatalogReloadRetryInterval, s.reloadRetry.interval)
 }
 
-func TestWithCatalogReloadRetryPolicy_OverridesDefaults(t *testing.T) {
-	t.Parallel()
-
-	s := NewDistributionServer(
-		distribution.NewEngine(),
-		distribution.NewCatalogStore(store.NewMVCCStore()),
-		WithCatalogReloadRetryPolicy(3, 5*time.Millisecond),
-	)
-	require.Equal(t, 3, s.reloadRetry.attempts)
-	require.Equal(t, 5*time.Millisecond, s.reloadRetry.interval)
-}
-
 func TestDistributionServerListRoutes_ReadsDurableCatalog(t *testing.T) {
 	t.Parallel()
 
@@ -1474,12 +1464,8 @@ func TestDistributionServerSplitRange_RetriesCatalogReloadUntilVisible(t *testin
 	require.NoError(t, err)
 	require.Equal(t, uint64(2), resp.CatalogVersion)
 
-	select {
-	case applyErr := <-coordinator.asyncApplyDone:
-		require.NoError(t, applyErr)
-	case <-time.After(time.Second):
-		t.Fatal("timed out waiting for async split apply")
-	}
+	applyErr := testutil.Recv(t, coordinator.asyncApplyDone, time.Second, "timed out waiting for async split apply")
+	require.NoError(t, applyErr)
 }
 
 func TestDistributionServerSplitRange_RequiresCatalogLeaderWhenCoordinatorConfigured(t *testing.T) {
@@ -1969,22 +1955,13 @@ func (a *distributionTSOAllocator) RunLeaseRenewal(ctx context.Context) {
 	<-ctx.Done()
 }
 
-type distributionLeaderView struct {
-	addr string
-}
-
-func (distributionLeaderView) State() raftengine.State { return raftengine.StateFollower }
-
-func (v distributionLeaderView) Leader() raftengine.LeaderInfo {
-	return raftengine.LeaderInfo{ID: "leader", Address: v.addr}
-}
-
-func (distributionLeaderView) VerifyLeader(context.Context) error {
-	return raftengine.ErrNotLeader
-}
-
-func (distributionLeaderView) LinearizableRead(context.Context) (uint64, error) {
-	return 0, raftengine.ErrNotLeader
+// distributionLeaderView reports a follower whose leader is at addr.
+func distributionLeaderView(addr string) *raftenginetest.StaticLeaderView {
+	return &raftenginetest.StaticLeaderView{
+		StateValue:  raftengine.StateFollower,
+		LeaderValue: raftengine.LeaderInfo{ID: "leader", Address: addr},
+		ReadErr:     raftengine.ErrNotLeader,
+	}
 }
 
 type recordingDistributionFilesystemObserver struct {

adapter/dynamodb_lease_validation_test.go

@@ -151,46 +151,6 @@ func TestDynamoDB_Query_MissingTableReturns400UnderDegradedLease(t *testing.T) {
 	require.Contains(t, rec.Body.String(), dynamoErrResourceNotFound)
 }
 
-// TestDynamoDB_LeaseCheckQuery_WholeTableQueryStillFencesAllGroups asserts the
-// fix preserves the all-groups fence for a VALID multi-shard read: a query whose
-// hash key is not the primary hash key reads the whole-table prefix (multiple
-// shards), so it must still fail closed when the lease cannot be confirmed.
-func TestDynamoDB_LeaseCheckQuery_WholeTableQueryStillFencesAllGroups(t *testing.T) {
-	t.Parallel()
-
-	st := store.NewMVCCStore()
-	writer := newDynamoFixtureWriter(t, st)
-	writer.writeSchema(leaseGSIFixtureSchema())
-
-	coord := newMultiGroupLeaseCoordinator()
-	server := NewDynamoDBServer(nil, st, coord)
-
-	// IndexName set + a VALID GSI on an EXISTING table => valid multi-shard
-	// read => keyless all-groups fence, not a skip. (Whole-table base-table
-	// queries take the same all-groups path through queryLeasePrefix's HashKey
-	// mismatch branch.) The schema must exist so the GSI is genuinely valid:
-	// after the P2-B fix an indexed query against a missing table or unknown
-	// index is classified as a skip, not an all-groups fence.
-	in := queryInput{
-		TableName:              "t",
-		IndexName:              "gsi1",
-		KeyConditionExpression: "gk = :gk",
-		ExpressionAttributeValues: map[string]attributeValue{
-			":gk": newStringAttributeValue("x"),
-		},
-	}
-
-	rec := httptest.NewRecorder()
-	req := httptest.NewRequest(http.MethodPost, "/", bytes.NewReader(nil))
-	require.True(t, server.leaseCheckQuery(rec, req, in))
-
-	fenced := coord.fencedGroupSet()
-	require.Contains(t, fenced, multiGroupDefaultID,
-		"valid multi-shard query must still fence the default group")
-	require.Contains(t, fenced, multiGroupOtherID,
-		"valid multi-shard query must still fence every group (must not be skipped as a validation case)")
-}
-
 // leaseGSIFixtureSchema is a table with a single GSI ("gsi1", ALL projection)
 // used by the lease pre-pass tests that exercise VALID multi-shard GSI reads.
 func leaseGSIFixtureSchema() *dynamoTableSchema {