migration: complete target promotion catalog state

.github/workflows/go-test.yml

@@ -12,6 +12,8 @@ permissions:
   contents: read
 jobs:
   test:
+    env:
+      GRPC_SEQUENCE_TESTS: '^(Test_consistency_satisfy_write_after_read_sequence|Test_grpc_transaction)$'
     strategy:
       matrix:
         os: [ [ubuntu-latest] ]
@@ -27,7 +29,12 @@ jobs:
         run: sudo apt-get update && sudo apt-get install -y fuse3
       - name: Reject production imports of test support
         run: bash scripts/check-production-imports.sh
-      - run: go test -race ./...
+      - name: Run race tests outside full-length gRPC sequences
+        run: go test -race ./... -skip "$GRPC_SEQUENCE_TESTS"
         env:
           # Fail, rather than skip, the kernel-mount tests when FUSE is unusable.
           ELASTICKV_REQUIRE_FUSE: "1"
+      # These unchanged 10,000-iteration tests need a fresh package budget
+      # after the serial migration/snapshot acceptance tests have finished.
+      - name: Run full-length gRPC consistency sequences
+        run: go test -race ./adapter -run "$GRPC_SEQUENCE_TESTS"

adapter/distribution_job_read_gate_test.go

@@ -0,0 +1,38 @@
+package adapter
+
+import (
+	"context"
+	"testing"
+
+	"github.com/bootjp/elastickv/distribution"
+	pb "github.com/bootjp/elastickv/proto"
+	"github.com/bootjp/elastickv/store"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+func TestDistributionServerJobReadsHonorStartupGate(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+	catalog := distribution.NewCatalogStore(store.NewMVCCStore())
+	job := sampleDistributionSplitJob(13)
+	require.NoError(t, catalog.CreateSplitJob(ctx, job))
+	blocked := true
+	server := NewDistributionServer(nil, catalog, WithDistributionReadGate(func() bool { return blocked }))
+	_, err := server.GetSplitJob(ctx, &pb.GetSplitJobRequest{JobId: job.JobID})
+	require.Equal(t, codes.Unavailable, status.Code(err))
+	_, err = server.GetSplitJob(ctx, &pb.GetSplitJobRequest{JobId: 99})
+	require.Equal(t, codes.Unavailable, status.Code(err), "catch-up must not return a stale NotFound")
+	_, err = server.ListSplitJobs(ctx, &pb.ListSplitJobsRequest{})
+	require.Equal(t, codes.Unavailable, status.Code(err))
+	blocked = false
+	got, err := server.GetSplitJob(ctx, &pb.GetSplitJobRequest{JobId: job.JobID})
+	require.NoError(t, err)
+	require.Equal(t, job.JobID, got.GetJob().GetJobId())
+	listed, err := server.ListSplitJobs(ctx, &pb.ListSplitJobsRequest{})
+	require.NoError(t, err)
+	require.Len(t, listed.GetJobs(), 1)
+	_, err = server.GetSplitJob(ctx, &pb.GetSplitJobRequest{JobId: 99})
+	require.Equal(t, codes.NotFound, status.Code(err))
+}