store: add migration version import export

.github/workflows/golangci-lint.yml

@@ -18,7 +18,18 @@ jobs:
     steps:
       - name: Check out code into the Go module directory
         uses: actions/checkout@v7
+      - name: Set up Go
+        uses: actions/setup-go@v6
+        with:
+          go-version: "1.x"
+      - name: Install golangci-lint
+        run: |
+          curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b "$(go env GOPATH)/bin" v2.9.0
+          echo "$(go env GOPATH)/bin" >> "$GITHUB_PATH"
+      - name: Run golangci-lint
+        run: golangci-lint run --config=.golangci.yaml
       - name: golangci-lint
+        continue-on-error: true
         uses: reviewdog/action-golangci-lint@v2
         with:
           github_token: ${{ secrets.GITHUB_TOKEN }}
@@ -28,4 +39,4 @@ jobs:
           filter_mode: nofilter
           reporter: github-pr-review
           cache: false
-          fail_on_error: true
+          fail_on_error: false

.github/workflows/redis-proxy-docker.yml

@@ -45,13 +45,29 @@ jobs:
 
       - name: Docker metadata
         id: meta
-        uses: docker/metadata-action@v6
-        with:
-          images: ghcr.io/${{ github.repository }}/redis-proxy
-          tags: |
-            type=sha
-            type=ref,event=branch
-            type=raw,value=latest,enable={{is_default_branch}}
+        shell: bash
+        env:
+          IMAGE: ghcr.io/${{ github.repository }}/redis-proxy
+          DEFAULT_BRANCH: ${{ github.event.repository.default_branch }}
+        run: |
+          short_sha="${GITHUB_SHA::7}"
+          ref_tag="$(printf '%s' "$GITHUB_REF_NAME" | sed -E 's/[^A-Za-z0-9_.-]+/-/g; s/^-+//; s/-+$//')"
+          ref_tag="${ref_tag:0:128}"
+          {
+            echo "tags<<EOF"
+            echo "${IMAGE}:sha-${short_sha}"
+            if [[ "${GITHUB_EVENT_NAME}" != "pull_request" && -n "${ref_tag}" ]]; then
+              echo "${IMAGE}:${ref_tag}"
+            fi
+            if [[ "${GITHUB_REF_NAME}" == "${DEFAULT_BRANCH}" ]]; then
+              echo "${IMAGE}:latest"
+            fi
+            echo "EOF"
+            echo "labels<<EOF"
+            echo "org.opencontainers.image.source=${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY}"
+            echo "org.opencontainers.image.revision=${GITHUB_SHA}"
+            echo "EOF"
+          } >> "$GITHUB_OUTPUT"
 
       - name: Build and push
         uses: docker/build-push-action@v7

adapter/distribution_server.go

@@ -29,6 +29,7 @@ type DistributionServer struct {
 	watchInterval time.Duration
 	watchLeader   func() bool
 	fsObserver    DistributionFilesystemObserver
+	readBlocked   func() bool
 	reloadRetry   struct {
 		attempts int
 		interval time.Duration
@@ -65,6 +66,12 @@ func WithDistributionFilesystemObserver(observer DistributionFilesystemObserver)
 	}
 }
 
+func WithDistributionReadGate(blocked func() bool) DistributionServerOption {
+	return func(s *DistributionServer) {
+		s.readBlocked = blocked
+	}
+}
+
 // WithCatalogReloadRetryPolicy configures the retry policy used after split
 // commit when waiting for the local catalog snapshot to become visible.
 func WithCatalogReloadRetryPolicy(attempts int, interval time.Duration) DistributionServerOption {
@@ -136,13 +143,30 @@ func NewDistributionServer(e *distribution.Engine, catalog *distribution.Catalog
 	return s
 }
 
+func (s *DistributionServer) SetReadGate(blocked func() bool) {
+	if s != nil {
+		s.readBlocked = blocked
+	}
+}
+
+func (s *DistributionServer) requireReadReady() error {
+	if s != nil && s.readBlocked != nil && s.readBlocked() {
+		//nolint:wrapcheck // Preserve the gRPC status code for startup readers.
+		return status.Error(codes.Unavailable, "distribution startup has not completed")
+	}
+	return nil
+}
+
 // UpdateRoute allows updating route information.
 func (s *DistributionServer) UpdateRoute(start, end []byte, group uint64) {
 	s.engine.UpdateRoute(start, end, group)
 }
 
 // GetRoute returns route for a key.
 func (s *DistributionServer) GetRoute(ctx context.Context, req *pb.GetRouteRequest) (*pb.GetRouteResponse, error) {
+	if err := s.requireReadReady(); err != nil {
+		return nil, err
+	}
 	r, ok := s.engine.GetRoute(kv.RouteKey(req.Key))
 	if !ok {
 		return &pb.GetRouteResponse{}, nil
@@ -162,6 +186,9 @@ func (s *DistributionServer) GetTimestamp(ctx context.Context, req *pb.GetTimest
 
 // ListRoutes returns all durable routes from catalog storage.
 func (s *DistributionServer) ListRoutes(ctx context.Context, req *pb.ListRoutesRequest) (*pb.ListRoutesResponse, error) {
+	if err := s.requireReadReady(); err != nil {
+		return nil, err
+	}
 	snapshot, err := s.loadCatalogSnapshot(ctx)
 	if err != nil {
 		return nil, err
@@ -352,32 +379,19 @@ func (s *DistributionServer) SplitRange(ctx context.Context, req *pb.SplitRangeR
 		return nil, err
 	}
 
-	parent, found := findRouteByID(snapshot.Routes, req.GetRouteId())
-	if !found {
-		return nil, grpcStatusError(codes.NotFound, errDistributionUnknownRoute.Error())
-	}
-
-	rawSplitKey := req.GetSplitKey()
-	splitKey := distribution.CloneBytes(fskeys.NormalizeSplitBoundary(kv.RouteKey(rawSplitKey)))
-	if err := validateSplitKey(parent, splitKey); err != nil {
-		s.observeFilePinnedHotspotIfNeeded(rawSplitKey, splitKey, err)
-		return nil, err
-	}
-
-	leftID, rightID, err := s.allocateChildRouteIDs(ctx, snapshot.ReadTS, snapshot.Routes)
+	plan, err := s.planSplitRange(ctx, snapshot, req)
 	if err != nil {
 		return nil, err
 	}
-	left, right := splitCatalogRoutes(parent, splitKey, leftID, rightID, 0)
 
-	saved, err := s.saveSplitResultViaCoordinator(ctx, snapshot.ReadTS, req.GetExpectedCatalogVersion(), parent.RouteID, left, right)
+	saved, err := s.saveSplitResultViaCoordinator(ctx, snapshot.ReadTS, req.GetExpectedCatalogVersion(), plan.parentID, plan.readKeys, plan.left, plan.right)
 	if err != nil {
 		return nil, err
 	}
 	if err := s.applyEngineSnapshot(saved); err != nil {
 		return nil, err
 	}
-	savedLeft, savedRight, err := splitChildrenFromSnapshot(saved, left.RouteID, right.RouteID)
+	savedLeft, savedRight, err := splitChildrenFromSnapshot(saved, plan.left.RouteID, plan.right.RouteID)
 	if err != nil {
 		return nil, err
 	}
@@ -389,6 +403,37 @@ func (s *DistributionServer) SplitRange(ctx context.Context, req *pb.SplitRangeR
 	}, nil
 }
 
+type splitRangePlan struct {
+	parentID uint64
+	readKeys [][]byte
+	left     distribution.RouteDescriptor
+	right    distribution.RouteDescriptor
+}
+
+func (s *DistributionServer) planSplitRange(ctx context.Context, snapshot distribution.CatalogSnapshot, req *pb.SplitRangeRequest) (splitRangePlan, error) {
+	parent, found := findRouteByID(snapshot.Routes, req.GetRouteId())
+	if !found {
+		return splitRangePlan{}, grpcStatusError(codes.NotFound, errDistributionUnknownRoute.Error())
+	}
+
+	rawSplitKey := req.GetSplitKey()
+	splitKey := distribution.CloneBytes(fskeys.NormalizeSplitBoundary(kv.RouteKey(rawSplitKey)))
+	if err := validateSplitKey(parent, splitKey); err != nil {
+		s.observeFilePinnedHotspotIfNeeded(rawSplitKey, splitKey, err)
+		return splitRangePlan{}, err
+	}
+	readKeys, err := s.splitJobOverlapReadKeys(ctx, snapshot, parent)
+	if err != nil {
+		return splitRangePlan{}, err
+	}
+	leftID, rightID, err := s.allocateChildRouteIDs(ctx, snapshot.ReadTS, snapshot.Routes)
+	if err != nil {
+		return splitRangePlan{}, err
+	}
+	left, right := splitCatalogRoutes(parent, splitKey, leftID, rightID, 0)
+	return splitRangePlan{parentID: parent.RouteID, readKeys: readKeys, left: left, right: right}, nil
+}
+
 func (s *DistributionServer) pinReadTS(ts uint64) *kv.ActiveTimestampToken {
 	if s == nil || s.readTracker == nil {
 		return nil
@@ -437,17 +482,14 @@ func (s *DistributionServer) saveSplitResultViaCoordinator(
 	readTS uint64,
 	expectedVersion uint64,
 	parentID uint64,
+	readKeys [][]byte,
 	left distribution.RouteDescriptor,
 	right distribution.RouteDescriptor,
 ) (distribution.CatalogSnapshot, error) {
-	if expectedVersion == math.MaxUint64 {
-		return distribution.CatalogSnapshot{}, grpcStatusError(codes.Internal, "catalog version overflow")
-	}
-	nextVersion := expectedVersion + 1
-	if right.RouteID == math.MaxUint64 {
-		return distribution.CatalogSnapshot{}, grpcStatusError(codes.Internal, errDistributionRouteIDOverflow.Error())
+	nextVersion, nextRouteID, err := nextCatalogSplitIDs(expectedVersion, right.RouteID)
+	if err != nil {
+		return distribution.CatalogSnapshot{}, err
 	}
-	nextRouteID := right.RouteID + 1
 	commitTS, err := kv.NextTimestampAfterThrough(ctx, s.coordinator, readTS, "split range: allocate commitTS")
 	if err != nil {
 		return distribution.CatalogSnapshot{}, grpcStatusErrorf(codes.Internal, "allocate split commit timestamp: %v", err)
@@ -482,8 +524,12 @@ func (s *DistributionServer) saveSplitResultViaCoordinator(
 		IsTxn:    true,
 		StartTS:  readTS,
 		CommitTS: commitTS,
+		ReadKeys: readKeys,
 	})
 	if err != nil {
+		if errors.Is(err, store.ErrWriteConflict) {
+			return distribution.CatalogSnapshot{}, grpcStatusError(codes.Aborted, errDistributionCatalogConflict.Error())
+		}
 		return distribution.CatalogSnapshot{}, grpcStatusErrorf(codes.Internal, "commit split mutations: %v", err)
 	}
 	if resp == nil || resp.CommitTS == 0 {
@@ -492,6 +538,16 @@ func (s *DistributionServer) saveSplitResultViaCoordinator(
 	return s.loadCatalogSnapshotAtVersion(ctx, resp.CommitTS, nextVersion)
 }
 
+func nextCatalogSplitIDs(expectedVersion uint64, rightRouteID uint64) (uint64, uint64, error) {
+	if expectedVersion == math.MaxUint64 {
+		return 0, 0, grpcStatusError(codes.Internal, "catalog version overflow")
+	}
+	if rightRouteID == math.MaxUint64 {
+		return 0, 0, grpcStatusError(codes.Internal, errDistributionRouteIDOverflow.Error())
+	}
+	return expectedVersion + 1, rightRouteID + 1, nil
+}
+
 func catalogStoreMutationsToOps(mutations []*store.KVPairMutation) ([]*kv.Elem[kv.OP], error) {
 	ops := make([]*kv.Elem[kv.OP], 0, len(mutations))
 	for _, mutation := range mutations {
@@ -672,6 +728,81 @@ func validateSplitKey(parent distribution.RouteDescriptor, splitKey []byte) erro
 	return nil
 }
 
+func (s *DistributionServer) splitJobOverlapReadKeys(ctx context.Context, snapshot distribution.CatalogSnapshot, parent distribution.RouteDescriptor) ([][]byte, error) {
+	jobs, err := s.catalog.ListSplitJobsAt(ctx, snapshot.ReadTS)
+	if err != nil {
+		return nil, grpcStatusErrorf(codes.Internal, "load split jobs: %v", err)
+	}
+	readKeys := splitJobReadFenceKeys(jobs)
+	for _, job := range jobs {
+		if !splitJobIsLive(job) {
+			continue
+		}
+		for _, interval := range liveSplitJobIntervals(job, snapshot.Routes) {
+			if routeRangeIntersects(parent.Start, parent.End, interval.start, interval.end) {
+				return nil, grpcStatusError(codes.Aborted, distribution.ErrSplitJobOverlap.Error())
+			}
+		}
+	}
+	return readKeys, nil
+}
+
+func splitJobReadFenceKeys(jobs []distribution.SplitJob) [][]byte {
+	readKeys := make([][]byte, 0, len(jobs)+1)
+	readKeys = append(readKeys, distribution.CatalogNextSplitJobIDKey())
+	for _, job := range jobs {
+		if splitJobIsLive(job) {
+			readKeys = append(readKeys, distribution.CatalogSplitJobKey(job.JobID))
+		}
+	}
+	return readKeys
+}
+
+func splitJobIsLive(job distribution.SplitJob) bool {
+	return job.Phase != distribution.SplitJobPhaseDone && job.Phase != distribution.SplitJobPhaseAbandoned
+}
+
+type routeInterval struct {
+	start []byte
+	end   []byte
+}
+
+const initialLiveSplitJobIntervalCapacity = 2
+
+func liveSplitJobIntervals(job distribution.SplitJob, routes []distribution.RouteDescriptor) []routeInterval {
+	out := make([]routeInterval, 0, initialLiveSplitJobIntervalCapacity)
+	for _, route := range routes {
+		switch {
+		case route.RouteID == job.SourceRouteID:
+			out = append(out, routeInterval{
+				start: distribution.CloneBytes(job.SplitKey),
+				end:   distribution.CloneBytes(route.End),
+			})
+		case route.ParentRouteID == job.SourceRouteID && routeRangeIntersects(route.Start, route.End, job.SplitKey, nil):
+			out = append(out, routeInterval{
+				start: distribution.CloneBytes(route.Start),
+				end:   distribution.CloneBytes(route.End),
+			})
+		case job.JobID != 0 && route.MigrationJobID == job.JobID:
+			out = append(out, routeInterval{
+				start: distribution.CloneBytes(route.Start),
+				end:   distribution.CloneBytes(route.End),
+			})
+		}
+	}
+	return out
+}
+
+func routeRangeIntersects(aStart, aEnd, bStart, bEnd []byte) bool {
+	if aEnd != nil && bytes.Compare(aEnd, bStart) <= 0 {
+		return false
+	}
+	if bEnd != nil && bytes.Compare(bEnd, aStart) <= 0 {
+		return false
+	}
+	return true
+}
+
 func splitCatalogRoutes(
 	parent distribution.RouteDescriptor,
 	splitKey []byte,
@@ -687,10 +818,10 @@ func splitCatalogRoutes(
 		GroupID:                parent.GroupID,
 		State:                  parent.State,
 		ParentRouteID:          parent.RouteID,
-		SplitAtHLC:             splitAtHLC,
 		StagedVisibilityActive: parent.StagedVisibilityActive,
 		MigrationJobID:         parent.MigrationJobID,
 		MinWriteTSExclusive:    parent.MinWriteTSExclusive,
+		SplitAtHLC:             splitAtHLC,
 	}
 	right := distribution.RouteDescriptor{
 		RouteID:                rightID,
@@ -699,10 +830,10 @@ func splitCatalogRoutes(
 		GroupID:                parent.GroupID,
 		State:                  parent.State,
 		ParentRouteID:          parent.RouteID,
-		SplitAtHLC:             splitAtHLC,
 		StagedVisibilityActive: parent.StagedVisibilityActive,
 		MigrationJobID:         parent.MigrationJobID,
 		MinWriteTSExclusive:    parent.MinWriteTSExclusive,
+		SplitAtHLC:             splitAtHLC,
 	}
 	return left, right
 }
@@ -758,10 +889,10 @@ func toProtoRouteDescriptor(route distribution.RouteDescriptor) *pb.RouteDescrip
 		RaftGroupId:            route.GroupID,
 		State:                  toProtoRouteState(route.State),
 		ParentRouteId:          route.ParentRouteID,
-		SplitAtHlc:             route.SplitAtHLC,
 		StagedVisibilityActive: route.StagedVisibilityActive,
 		MigrationJobId:         route.MigrationJobID,
 		MinWriteTsExclusive:    route.MinWriteTSExclusive,
+		SplitAtHlc:             route.SplitAtHLC,
 	}
 }