migration: complete target promotion catalog state

adapter/internal_migration_probe_test.go

@@ -0,0 +1,154 @@
+package adapter
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/bootjp/elastickv/distribution"
+	"github.com/bootjp/elastickv/kv"
+	pb "github.com/bootjp/elastickv/proto"
+	"github.com/bootjp/elastickv/store"
+	"github.com/stretchr/testify/require"
+)
+
+func TestInternalProbeMigrationStateUsesLocalFSMAndCatalog(t *testing.T) {
+	t.Parallel()
+
+	ctx := context.Background()
+	st := store.NewMVCCStore()
+	engine := distribution.NewEngine()
+	require.NoError(t, engine.ApplySnapshot(distribution.CatalogSnapshot{
+		Version: 9,
+		Routes: []distribution.RouteDescriptor{{
+			RouteID:             2,
+			Start:               []byte("m"),
+			GroupID:             2,
+			State:               distribution.RouteStateActive,
+			MinWriteTSExclusive: 88,
+		}},
+	}))
+	tracker := kv.NewActiveTimestampTracker()
+	internal := NewInternalWithEngine(nil, nil, nil, nil,
+		WithInternalStore(st),
+		WithInternalRouteEngine(engine),
+		WithInternalActiveTimestampTracker(tracker),
+	)
+
+	state := store.TargetStagedReadinessState{
+		JobID:                  7,
+		RouteStart:             []byte("m"),
+		ExpectedCutoverVersion: 9,
+		MigrationJobID:         7,
+		MinWriteTSExclusive:    88,
+		Armed:                  true,
+	}
+	readinessWriter, ok := st.(store.MigrationTargetReadinessWriter)
+	require.True(t, ok)
+	require.NoError(t, readinessWriter.ApplyTargetStagedReadiness(ctx, state))
+	control, err := internal.ProbeMigrationState(ctx, &pb.ProbeMigrationStateRequest{
+		JobId:                  7,
+		Kind:                   pb.MigrationStateProbeKind_MIGRATION_STATE_PROBE_KIND_CONTROL_APPLIED,
+		RouteStart:             []byte("m"),
+		ExpectedCatalogVersion: 9,
+		MigrationJobId:         7,
+		MinWriteTsExclusive:    88,
+	})
+	require.NoError(t, err)
+	require.True(t, control.Ready)
+
+	cleared, err := internal.ProbeMigrationState(ctx, &pb.ProbeMigrationStateRequest{
+		JobId:                  7,
+		Kind:                   pb.MigrationStateProbeKind_MIGRATION_STATE_PROBE_KIND_TARGET_DESCRIPTOR_CLEARED,
+		RouteStart:             []byte("m"),
+		ExpectedCatalogVersion: 9,
+		ExpectedGroupId:        2,
+		MinWriteTsExclusive:    88,
+	})
+	require.NoError(t, err)
+	require.True(t, cleared.Ready)
+
+	pin := tracker.Pin(100)
+	drained, err := internal.ProbeMigrationState(ctx, &pb.ProbeMigrationStateRequest{
+		JobId:                7,
+		Kind:                 pb.MigrationStateProbeKind_MIGRATION_STATE_PROBE_KIND_SOURCE_READ_DRAINED,
+		ReadDrainNotBeforeMs: time.Now().Add(-time.Second).UnixMilli(),
+	})
+	require.NoError(t, err)
+	require.False(t, drained.Ready)
+	pin.Release()
+	drained, err = internal.ProbeMigrationState(ctx, &pb.ProbeMigrationStateRequest{
+		JobId:                7,
+		Kind:                 pb.MigrationStateProbeKind_MIGRATION_STATE_PROBE_KIND_SOURCE_READ_DRAINED,
+		ReadDrainNotBeforeMs: time.Now().Add(-time.Second).UnixMilli(),
+	})
+	require.NoError(t, err)
+	require.True(t, drained.Ready)
+
+	metadata, err := internal.ProbeMigrationState(ctx, &pb.ProbeMigrationStateRequest{
+		JobId: 7,
+		Kind:  pb.MigrationStateProbeKind_MIGRATION_STATE_PROBE_KIND_METADATA_CLEARED,
+	})
+	require.NoError(t, err)
+	require.False(t, metadata.Ready)
+	cleaner, ok := st.(store.MigrationCleaner)
+	require.True(t, ok)
+	require.NoError(t, cleaner.ClearMigrationState(ctx, 7, 0))
+	metadata, err = internal.ProbeMigrationState(ctx, &pb.ProbeMigrationStateRequest{
+		JobId: 7,
+		Kind:  pb.MigrationStateProbeKind_MIGRATION_STATE_PROBE_KIND_METADATA_CLEARED,
+	})
+	require.NoError(t, err)
+	require.True(t, metadata.Ready)
+}
+
+func TestInternalProbeMigrationMetadataClearedWaitsForImportMetadata(t *testing.T) {
+	t.Parallel()
+
+	ctx := context.Background()
+	st := store.NewMVCCStore()
+	internal := NewInternalWithEngine(nil, nil, nil, nil, WithInternalStore(st))
+	_, err := st.ImportVersions(ctx, store.ImportVersionsOptions{
+		JobID:     7,
+		BracketID: 1,
+		BatchSeq:  1,
+		Cursor:    []byte("cursor-1"),
+		Versions: []store.MVCCVersion{{
+			Key:      []byte("m/key"),
+			Value:    []byte("value"),
+			CommitTS: 42,
+		}},
+	})
+	require.NoError(t, err)
+
+	metadata, err := internal.ProbeMigrationState(ctx, &pb.ProbeMigrationStateRequest{
+		JobId: 7,
+		Kind:  pb.MigrationStateProbeKind_MIGRATION_STATE_PROBE_KIND_METADATA_CLEARED,
+	})
+	require.NoError(t, err)
+	require.False(t, metadata.Ready)
+
+	cleaner, ok := st.(store.MigrationCleaner)
+	require.True(t, ok)
+	require.NoError(t, cleaner.ClearMigrationState(ctx, 7, 0))
+	metadata, err = internal.ProbeMigrationState(ctx, &pb.ProbeMigrationStateRequest{
+		JobId: 7,
+		Kind:  pb.MigrationStateProbeKind_MIGRATION_STATE_PROBE_KIND_METADATA_CLEARED,
+	})
+	require.NoError(t, err)
+	require.True(t, metadata.Ready)
+}
+
+func TestInternalIssueMigrationTimestampFollowsSourceLastCommit(t *testing.T) {
+	t.Parallel()
+
+	ctx := context.Background()
+	st := store.NewMVCCStore()
+	require.NoError(t, st.ApplyMutations(ctx, []*store.KVPairMutation{{Key: []byte("m"), Value: []byte("value")}}, nil, 50, 50))
+	internal := NewInternalWithEngine(nil, mockInternalLeader{}, nil, nil, WithInternalStore(st))
+
+	resp, err := internal.IssueMigrationTimestamp(ctx, &pb.IssueMigrationTimestampRequest{})
+	require.NoError(t, err)
+	require.Equal(t, uint64(50), resp.GetLastCommitTs())
+	require.Greater(t, resp.GetTimestamp(), resp.GetLastCommitTs())
+}

adapter/internal_migration_test.go

@@ -659,3 +659,137 @@ func TestInternalPromoteStagedVersionsAppliesStoreBatch(t *testing.T) {
 	require.ErrorIs(t, err, store.ErrKeyNotFound)
 	require.GreaterOrEqual(t, clock.Current(), uint64(30))
 }
+
+func TestInternalApplyTargetStagedReadinessBypassesOpcodeGateForSafetyGuard(t *testing.T) {
+	t.Parallel()
+
+	ctx := context.Background()
+	st := store.NewMVCCStore()
+	proposer := &applyingMigrationProposer{
+		fsm: kv.NewKvFSMWithHLC(st, nil),
+	}
+	internal := NewInternalWithEngine(nil, mockInternalLeader{}, nil, nil,
+		WithInternalStore(st),
+		WithInternalMigrationProposer(proposer),
+		WithInternalMigrationPromoteGate(func(context.Context) error {
+			return status.Error(codes.FailedPrecondition, "migration opcode disabled for test")
+		}),
+	)
+
+	resp, err := internal.ApplyTargetStagedReadiness(ctx, &pb.TargetStagedReadinessRequest{
+		JobId:                  9,
+		RouteStart:             []byte("a"),
+		RouteEnd:               []byte("z"),
+		ExpectedCutoverVersion: 3,
+		MigrationJobId:         7,
+		MinWriteTsExclusive:    100,
+		Armed:                  true,
+	})
+	require.NoError(t, err)
+	require.NotNil(t, resp)
+	require.Equal(t, uint64(1), proposer.calls)
+
+	reader, ok := st.(store.MigrationTargetReadinessReader)
+	require.True(t, ok)
+	states, err := reader.MigrationTargetReadinessStates(ctx)
+	require.NoError(t, err)
+	require.Equal(t, []store.TargetStagedReadinessState{{
+		JobID:                  9,
+		RouteStart:             []byte("a"),
+		RouteEnd:               []byte("z"),
+		ExpectedCutoverVersion: 3,
+		MigrationJobID:         7,
+		MinWriteTSExclusive:    100,
+		Armed:                  true,
+	}}, states)
+}
+
+func TestInternalApplyTargetStagedReadinessProposesThroughRaft(t *testing.T) {
+	t.Parallel()
+
+	ctx := context.Background()
+	st := store.NewMVCCStore()
+	proposer := &applyingMigrationProposer{
+		fsm: kv.NewKvFSMWithHLC(st, nil),
+	}
+	internal := NewInternalWithEngine(nil, mockInternalLeader{}, nil, nil,
+		WithInternalStore(st),
+		WithInternalMigrationProposer(proposer),
+		WithInternalMigrationPromoteGate(func(context.Context) error { return nil }),
+	)
+
+	_, err := internal.ApplyTargetStagedReadiness(ctx, &pb.TargetStagedReadinessRequest{
+		JobId:                  9,
+		RouteStart:             []byte("a"),
+		RouteEnd:               []byte("z"),
+		ExpectedCutoverVersion: 3,
+		MigrationJobId:         7,
+		MinWriteTsExclusive:    100,
+		Armed:                  true,
+	})
+	require.NoError(t, err)
+	require.Equal(t, uint64(1), proposer.calls)
+
+	reader, ok := st.(store.MigrationTargetReadinessReader)
+	require.True(t, ok)
+	states, err := reader.MigrationTargetReadinessStates(ctx)
+	require.NoError(t, err)
+	require.Equal(t, []store.TargetStagedReadinessState{{
+		JobID:                  9,
+		RouteStart:             []byte("a"),
+		RouteEnd:               []byte("z"),
+		ExpectedCutoverVersion: 3,
+		MigrationJobID:         7,
+		MinWriteTSExclusive:    100,
+		Armed:                  true,
+	}}, states)
+}
+
+func TestInternalApplyTargetStagedReadinessRejectsArmedZeroMinWriteTS(t *testing.T) {
+	t.Parallel()
+
+	ctx := context.Background()
+	st := store.NewMVCCStore()
+	proposer := &applyingMigrationProposer{
+		fsm: kv.NewKvFSMWithHLC(st, nil),
+	}
+	internal := NewInternalWithEngine(nil, mockInternalLeader{}, nil, nil,
+		WithInternalStore(st),
+		WithInternalMigrationProposer(proposer),
+		WithInternalMigrationPromoteGate(func(context.Context) error { return nil }),
+	)
+
+	resp, err := internal.ApplyTargetStagedReadiness(ctx, &pb.TargetStagedReadinessRequest{
+		JobId:                  9,
+		RouteStart:             []byte("a"),
+		RouteEnd:               []byte("z"),
+		ExpectedCutoverVersion: 3,
+		MigrationJobId:         7,
+		Armed:                  true,
+	})
+	require.Nil(t, resp)
+	require.Error(t, err)
+	require.Equal(t, codes.InvalidArgument, status.Code(err))
+	require.Equal(t, uint64(0), proposer.calls)
+}
+
+func TestInternalCleanupMigrationRejectsWhenOpcodeGateClosed(t *testing.T) {
+	t.Parallel()
+
+	proposer := &applyingMigrationProposer{fsm: kv.NewKvFSMWithHLC(store.NewMVCCStore(), nil)}
+	internal := NewInternalWithEngine(nil, mockInternalLeader{}, nil, nil,
+		WithInternalMigrationProposer(proposer),
+		WithInternalMigrationCleanupGate(func(context.Context) error {
+			return status.Error(codes.FailedPrecondition, "migration cleanup disabled for test")
+		}),
+	)
+
+	resp, err := internal.CleanupMigration(context.Background(), &pb.CleanupMigrationRequest{
+		JobId:     9,
+		Mode:      pb.MigrationCleanupMode_MIGRATION_CLEANUP_MODE_VERSIONS,
+		KeyFamily: distribution.MigrationFamilyUser,
+	})
+	require.Nil(t, resp)
+	require.Equal(t, codes.FailedPrecondition, status.Code(err))
+	require.Zero(t, proposer.calls)
+}

adapter/redis_compat_commands_stream_test.go

@@ -279,19 +279,27 @@ func TestRedis_StreamXReadLatencyIsConstant(t *testing.T) {
 
 	rdb := redis.NewClient(&redis.Options{Addr: nodes[0].redisAddress})
 	defer func() { _ = rdb.Close() }()
-	ctx := context.Background()
+	ctx := t.Context()
 
 	const probes = 100
 	lastID := seedStreamEntriesForXReadLatency(t, nodes[0].redisServer, ctx, "stream-lat")
 
 	measure := func() time.Duration {
-		start := time.Now()
-		streams, err := rdb.XRead(ctx, &redis.XReadArgs{
-			Streams: []string{"stream-lat", lastID},
-			Count:   10,
-			Block:   10 * time.Millisecond,
-		}).Result()
-		elapsed := time.Since(start)
+		var (
+			streams []redis.XStream
+			elapsed time.Duration
+		)
+		err := retryNotLeader(ctx, func() error {
+			start := time.Now()
+			var xerr error
+			streams, xerr = rdb.XRead(ctx, &redis.XReadArgs{
+				Streams: []string{"stream-lat", lastID},
+				Count:   10,
+				Block:   10 * time.Millisecond,
+			}).Result()
+			elapsed = time.Since(start)
+			return xerr
+		})
 		require.True(t, errors.Is(err, redis.Nil) || err == nil)
 		require.Empty(t, streams)
 		return elapsed

adapter/redis_delta_compactor_test.go

@@ -1684,8 +1684,11 @@ func TestDeltaCompactor_UrgentCompactionPagination(t *testing.T) {
 		require.NoError(t, st.PutAt(ctx, dKey, delta, i+1, 0))
 	}
 
-	// Queue and process the urgent compaction.
-	c.TriggerUrgentCompaction("hash", userKey)
+	// Exercise the urgent pagination loop directly. Running through c.Run would
+	// also start an initial background SyncOnce; the local test coordinator applies
+	// elems one-by-one instead of atomically, so a test read can observe the meta
+	// update before all delete elems have been applied under the race detector.
+	c.compactUrgentKey(ctx, urgentCompactionRequest{typeName: "hash", userKey: userKey})
 
 	runCtx, cancel := context.WithCancel(ctx)
 	defer cancel()