diff --git a/adapter/distribution_server.go b/adapter/distribution_server.go index bc48b33fd..dafb3649b 100644 --- a/adapter/distribution_server.go +++ b/adapter/distribution_server.go @@ -3,11 +3,15 @@ package adapter import ( "bytes" "context" + "encoding/binary" "math" + "sort" + "strings" "sync" "time" "github.com/bootjp/elastickv/distribution" + "github.com/bootjp/elastickv/internal/raftengine" "github.com/bootjp/elastickv/kv" pb "github.com/bootjp/elastickv/proto" "github.com/bootjp/elastickv/store" @@ -18,12 +22,15 @@ import ( // DistributionServer serves distribution related gRPC APIs. type DistributionServer struct { - mu sync.Mutex - engine *distribution.Engine - catalog *distribution.CatalogStore - coordinator kv.Coordinator - readTracker *kv.ActiveTimestampTracker - reloadRetry struct { + mu sync.Mutex + engine *distribution.Engine + catalog *distribution.CatalogStore + coordinator kv.Coordinator + readTracker *kv.ActiveTimestampTracker + migrationCapabilityGate SplitMigrationCapabilityGate + splitJobRunnerReady bool + knownRaftGroups map[uint64]struct{} + reloadRetry struct { attempts int interval time.Duration } @@ -33,6 +40,10 @@ type DistributionServer struct { // DistributionServerOption configures DistributionServer behavior. type DistributionServerOption func(*DistributionServer) +// SplitMigrationCapabilityGate reports whether this node can safely create +// migration-only side effects. A nil gate keeps StartSplitMigration fail-closed. +type SplitMigrationCapabilityGate func(context.Context) error + // WithDistributionCoordinator configures the coordinator used for Raft-backed // catalog mutations in SplitRange. func WithDistributionCoordinator(coordinator kv.Coordinator) DistributionServerOption { @@ -47,6 +58,35 @@ func WithDistributionActiveTimestampTracker(tracker *kv.ActiveTimestampTracker) } } +func WithSplitMigrationCapabilityGate(gate SplitMigrationCapabilityGate) DistributionServerOption { + return func(s *DistributionServer) { + s.migrationCapabilityGate = gate + } +} + +// WithSplitJobRunnerReady opens the split migration capability probe once this +// node can advance planned split jobs. +func WithSplitJobRunnerReady() DistributionServerOption { + return func(s *DistributionServer) { + s.splitJobRunnerReady = true + } +} + +// WithDistributionKnownRaftGroups configures the Raft group IDs this node can +// route migration work to. +func WithDistributionKnownRaftGroups(groupIDs ...uint64) DistributionServerOption { + return func(s *DistributionServer) { + groups := make(map[uint64]struct{}, len(groupIDs)) + for _, groupID := range groupIDs { + if groupID == 0 { + continue + } + groups[groupID] = struct{}{} + } + s.knownRaftGroups = groups + } +} + // 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 { @@ -61,8 +101,15 @@ func WithCatalogReloadRetryPolicy(attempts int, interval time.Duration) Distribu } const ( - childRouteCount = 2 - splitMutationOpCount = childRouteCount + 3 + childRouteCount = 2 + splitMutationOpCount = childRouteCount + 3 + listSplitJobsDefaultPageSize = 200 + splitJobListCursorVersion = byte(1) + splitJobListCursorTerminalOff = 1 + splitJobListCursorJobIDOff = splitJobListCursorTerminalOff + 8 + splitJobListCursorEncodedBytes = splitJobListCursorJobIDOff + 8 + SplitMigrationCapabilityV2 = "cap_migration_v2" + splitMigrationCapabilityV2 = SplitMigrationCapabilityV2 ) var ( @@ -79,6 +126,10 @@ var ( errDistributionCoordinatorRequired = errors.New("distribution coordinator is not configured") errDistributionEngineNotConfigured = errors.New("distribution engine is not configured") errDistributionCatalogVersionNotFound = errors.New("route catalog version not found") + errDistributionClusterNotReady = errors.New("cluster is not ready for split migration") + errDistributionRaftGroupsNotKnown = errors.New("distribution raft groups are not configured") + errDistributionUnknownTargetGroup = errors.New("unknown target group") + errDistributionSourceRouteNotActive = errors.New("source route is not active") ) // NewDistributionServer creates a new server. @@ -173,6 +224,201 @@ func (s *DistributionServer) GetIntersectingRoutes(ctx context.Context, req *pb. }, nil } +func (s *DistributionServer) StartSplitMigration(ctx context.Context, req *pb.StartSplitMigrationRequest) (*pb.StartSplitMigrationResponse, error) { + if err := s.verifyStartSplitMigrationPreflight(ctx, req); err != nil { + return nil, err + } + + s.mu.Lock() + defer s.mu.Unlock() + + if err := s.verifyStartSplitMigrationCatalogReady(ctx); err != nil { + return nil, err + } + snapshot, err := s.loadCatalogSnapshot(ctx) + if err != nil { + return nil, err + } + defer releaseReadPin(s.pinReadTS(snapshot.ReadTS)) + + parent, err := s.startSplitMigrationParent(ctx, snapshot, req) + if err != nil { + return nil, err + } + job, err := s.newSplitMigrationJob(ctx, snapshot, parent, req) + if err != nil { + return nil, err + } + if err := s.createSplitJobViaCoordinator(ctx, snapshot.ReadTS, job); err != nil { + return nil, err + } + return &pb.StartSplitMigrationResponse{ + CatalogVersion: snapshot.Version, + JobId: job.JobID, + }, nil +} + +func (s *DistributionServer) GetSplitMigrationCapability(context.Context, *pb.GetSplitMigrationCapabilityRequest) (*pb.GetSplitMigrationCapabilityResponse, error) { + if !s.splitJobRunnerReady { + return &pb.GetSplitMigrationCapabilityResponse{}, nil + } + return &pb.GetSplitMigrationCapabilityResponse{ + MigrationCapable: true, + Capabilities: []string{splitMigrationCapabilityV2}, + }, nil +} + +func (s *DistributionServer) verifyStartSplitMigrationPreflight(ctx context.Context, req *pb.StartSplitMigrationRequest) error { + if req.GetTargetGroupId() == 0 { + return grpcStatusError(codes.InvalidArgument, distribution.ErrCatalogSplitJobTargetGroupRequired.Error()) + } + if err := s.verifyStartSplitMigrationCatalogReady(ctx); err != nil { + return err + } + return s.verifySplitMigrationCapability(ctx) +} + +func (s *DistributionServer) verifyStartSplitMigrationCatalogReady(ctx context.Context) error { + if err := s.verifyCatalogLeader(ctx); err != nil { + return err + } + if s.catalog == nil { + return grpcStatusError(codes.FailedPrecondition, errDistributionCatalogNotConfigured.Error()) + } + return nil +} + +func (s *DistributionServer) startSplitMigrationParent( + ctx context.Context, + snapshot distribution.CatalogSnapshot, + req *pb.StartSplitMigrationRequest, +) (distribution.RouteDescriptor, error) { + if err := validateExpectedCatalogVersion(snapshot.Version, req.GetExpectedCatalogVersion()); err != nil { + return distribution.RouteDescriptor{}, err + } + parent, found := findRouteByID(snapshot.Routes, req.GetRouteId()) + if !found { + return distribution.RouteDescriptor{}, grpcStatusError(codes.NotFound, errDistributionUnknownRoute.Error()) + } + if parent.State != distribution.RouteStateActive { + return distribution.RouteDescriptor{}, grpcStatusError(codes.FailedPrecondition, errDistributionSourceRouteNotActive.Error()) + } + splitKey := distribution.CloneBytes(req.GetSplitKey()) + if err := validateSplitKey(parent, splitKey); err != nil { + return distribution.RouteDescriptor{}, err + } + if err := distribution.ValidateMigrationRouteRange(splitKey, parent.End); err != nil { + return distribution.RouteDescriptor{}, splitMigrationRangeStatusError(err) + } + if parent.GroupID == req.GetTargetGroupId() { + return distribution.RouteDescriptor{}, grpcStatusError(codes.InvalidArgument, "target group must differ from source route group") + } + if err := s.verifyKnownTargetGroup(req.GetTargetGroupId()); err != nil { + return distribution.RouteDescriptor{}, err + } + if err := s.rejectLiveSplitJob(ctx, snapshot, parent); err != nil { + return distribution.RouteDescriptor{}, err + } + return parent, nil +} + +func (s *DistributionServer) newSplitMigrationJob( + ctx context.Context, + snapshot distribution.CatalogSnapshot, + parent distribution.RouteDescriptor, + req *pb.StartSplitMigrationRequest, +) (distribution.SplitJob, error) { + jobID, err := s.catalog.NextSplitJobIDAt(ctx, snapshot.ReadTS) + if err != nil { + return distribution.SplitJob{}, splitJobCatalogStatusError(err) + } + job, err := distribution.InitializeSplitJobPlan(distribution.SplitJob{ + JobID: jobID, + SourceRouteID: parent.RouteID, + SplitKey: distribution.CloneBytes(req.GetSplitKey()), + TargetGroupID: req.GetTargetGroupId(), + }, parent, time.Now().UnixMilli()) + if err != nil { + return distribution.SplitJob{}, splitJobCatalogStatusError(err) + } + return job, nil +} + +func (s *DistributionServer) GetSplitJob(ctx context.Context, req *pb.GetSplitJobRequest) (*pb.GetSplitJobResponse, error) { + if req.GetJobId() == 0 { + return nil, grpcStatusError(codes.InvalidArgument, distribution.ErrCatalogSplitJobIDRequired.Error()) + } + if s.catalog == nil { + return nil, grpcStatusError(codes.FailedPrecondition, errDistributionCatalogNotConfigured.Error()) + } + job, found, err := s.catalog.SplitJob(ctx, req.GetJobId()) + if err != nil { + return nil, splitJobCatalogStatusError(err) + } + if !found { + return nil, grpcStatusError(codes.NotFound, distribution.ErrCatalogSplitJobNotFound.Error()) + } + return &pb.GetSplitJobResponse{Job: distribution.SplitJobToProto(job)}, nil +} + +func (s *DistributionServer) ListSplitJobs(ctx context.Context, req *pb.ListSplitJobsRequest) (*pb.ListSplitJobsResponse, error) { + if s.catalog == nil { + return nil, grpcStatusError(codes.FailedPrecondition, errDistributionCatalogNotConfigured.Error()) + } + if req == nil { + req = &pb.ListSplitJobsRequest{} + } + phaseFilter := normalizeSplitJobPhaseFilter(req.GetPhase()) + if phaseFilter != "" && !validSplitJobPhaseFilter(phaseFilter) { + return nil, grpcStatusErrorf(codes.InvalidArgument, "unknown split job phase %q", req.GetPhase()) + } + jobs, err := s.catalog.ListSplitJobs(ctx) + if err != nil { + return nil, splitJobCatalogStatusError(err) + } + resp, err := splitJobListPage(jobs, req, phaseFilter) + if err != nil { + return nil, err + } + return resp, nil +} + +func (s *DistributionServer) AbandonSplitJob(ctx context.Context, req *pb.AbandonSplitJobRequest) (*pb.AbandonSplitJobResponse, error) { + if req.GetJobId() == 0 { + return nil, grpcStatusError(codes.InvalidArgument, distribution.ErrCatalogSplitJobIDRequired.Error()) + } + if err := s.verifyCatalogLeader(ctx); err != nil { + return nil, err + } + if s.catalog == nil { + return nil, grpcStatusError(codes.FailedPrecondition, errDistributionCatalogNotConfigured.Error()) + } + if err := s.updateSplitJobViaCoordinator(ctx, req.GetJobId(), func(job distribution.SplitJob) (distribution.SplitJob, error) { + return distribution.BeginSplitJobAbandon(job, time.Now().UnixMilli()) + }); err != nil { + return nil, err + } + return &pb.AbandonSplitJobResponse{}, nil +} + +func (s *DistributionServer) RetrySplitJob(ctx context.Context, req *pb.RetrySplitJobRequest) (*pb.RetrySplitJobResponse, error) { + if req.GetJobId() == 0 { + return nil, grpcStatusError(codes.InvalidArgument, distribution.ErrCatalogSplitJobIDRequired.Error()) + } + if err := s.verifyCatalogLeader(ctx); err != nil { + return nil, err + } + if s.catalog == nil { + return nil, grpcStatusError(codes.FailedPrecondition, errDistributionCatalogNotConfigured.Error()) + } + if err := s.updateSplitJobViaCoordinator(ctx, req.GetJobId(), func(job distribution.SplitJob) (distribution.SplitJob, error) { + return distribution.RetrySplitJobState(job, time.Now().UnixMilli()) + }); err != nil { + return nil, err + } + return &pb.RetrySplitJobResponse{}, nil +} + func (s *DistributionServer) routeSnapshotAt(version uint64) (distribution.RouteHistorySnapshot, error) { if s.engine == nil { return distribution.RouteHistorySnapshot{}, grpcStatusError(codes.FailedPrecondition, errDistributionEngineNotConfigured.Error()) @@ -199,13 +445,12 @@ func (s *DistributionServer) SplitRange(ctx context.Context, req *pb.SplitRangeR if err != nil { return nil, err } - readPin := s.pinReadTS(snapshot.ReadTS) - defer readPin.Release() + defer releaseReadPin(s.pinReadTS(snapshot.ReadTS)) if err := validateExpectedCatalogVersion(snapshot.Version, req.GetExpectedCatalogVersion()); err != nil { return nil, err } - parent, _, found := findRouteByID(snapshot.Routes, req.GetRouteId()) + parent, found := findRouteByID(snapshot.Routes, req.GetRouteId()) if !found { return nil, grpcStatusError(codes.NotFound, errDistributionUnknownRoute.Error()) } @@ -242,11 +487,18 @@ func (s *DistributionServer) SplitRange(ctx context.Context, req *pb.SplitRangeR func (s *DistributionServer) pinReadTS(ts uint64) *kv.ActiveTimestampToken { if s == nil || s.readTracker == nil { - return nil + return &kv.ActiveTimestampToken{} } return s.readTracker.Pin(ts) } +func releaseReadPin(token *kv.ActiveTimestampToken) { + if token == nil { + return + } + token.Release() +} + func (s *DistributionServer) verifyCatalogLeader(ctx context.Context) error { if s.coordinator == nil { return grpcStatusError(codes.FailedPrecondition, errDistributionCoordinatorRequired.Error()) @@ -409,6 +661,99 @@ func (s *DistributionServer) applyEngineSnapshot(snapshot distribution.CatalogSn return nil } +func (s *DistributionServer) updateSplitJobViaCoordinator( + ctx context.Context, + jobID uint64, + transition func(distribution.SplitJob) (distribution.SplitJob, error), +) error { + expected, readTS, err := s.catalog.LiveSplitJobForUpdate(ctx, jobID) + if err != nil { + return splitJobCatalogStatusError(err) + } + next, err := transition(expected) + if err != nil { + return splitJobCatalogStatusError(err) + } + if distribution.SplitJobsEquivalent(expected, next) { + return nil + } + encoded, err := distribution.EncodeSplitJob(next) + if err != nil { + return splitJobCatalogStatusError(err) + } + key := distribution.CatalogSplitJobKey(jobID) + if _, err := s.coordinator.Dispatch(ctx, &kv.OperationGroup[kv.OP]{ + Elems: []*kv.Elem[kv.OP]{{ + Op: kv.Put, + Key: key, + Value: encoded, + }}, + IsTxn: true, + StartTS: readTS, + ReadKeys: [][]byte{key}, + }); err != nil { + return splitJobCoordinatorStatusError(err) + } + return nil +} + +func (s *DistributionServer) verifyKnownTargetGroup(groupID uint64) error { + if len(s.knownRaftGroups) == 0 { + return grpcStatusError(codes.FailedPrecondition, errDistributionRaftGroupsNotKnown.Error()) + } + if _, ok := s.knownRaftGroups[groupID]; !ok { + return grpcStatusError(codes.InvalidArgument, errDistributionUnknownTargetGroup.Error()) + } + return nil +} + +func (s *DistributionServer) createSplitJobViaCoordinator(ctx context.Context, readTS uint64, job distribution.SplitJob) error { + if job.JobID == math.MaxUint64 { + return splitJobCatalogStatusError(distribution.ErrCatalogSplitJobIDOverflow) + } + encoded, err := distribution.EncodeSplitJob(job) + if err != nil { + return splitJobCatalogStatusError(err) + } + jobKey := distribution.CatalogSplitJobKey(job.JobID) + nextIDKey := distribution.CatalogNextSplitJobIDKey() + if _, err := s.coordinator.Dispatch(ctx, &kv.OperationGroup[kv.OP]{ + Elems: []*kv.Elem[kv.OP]{ + { + Op: kv.Put, + Key: jobKey, + Value: encoded, + }, + { + Op: kv.Put, + Key: nextIDKey, + Value: distribution.EncodeCatalogNextSplitJobID(job.JobID + 1), + }, + }, + IsTxn: true, + StartTS: readTS, + ReadKeys: [][]byte{ + jobKey, + nextIDKey, + distribution.CatalogVersionKey(), + distribution.CatalogRouteKey(job.SourceRouteID), + }, + }); err != nil { + return splitJobCoordinatorStatusError(err) + } + return nil +} + +func (s *DistributionServer) verifySplitMigrationCapability(ctx context.Context) error { + if s.migrationCapabilityGate == nil { + return grpcStatusError(codes.FailedPrecondition, errDistributionClusterNotReady.Error()) + } + if err := s.migrationCapabilityGate(ctx); err != nil { + return err + } + return nil +} + func validateExpectedCatalogVersion(currentVersion, expectedVersion uint64) error { if currentVersion != expectedVersion { return grpcStatusError(codes.Aborted, errDistributionCatalogConflict.Error()) @@ -470,6 +815,29 @@ func splitJobReadFenceKeys(jobs []distribution.SplitJob) [][]byte { return readKeys } +func (s *DistributionServer) rejectLiveSplitJob(ctx context.Context, snapshot distribution.CatalogSnapshot, parent distribution.RouteDescriptor) error { + jobs, err := s.catalog.ListSplitJobsAt(ctx, snapshot.ReadTS) + if err != nil { + return grpcStatusErrorf(codes.Internal, "load split jobs: %v", err) + } + liveJobs := 0 + for _, job := range jobs { + if !splitJobIsLive(job) { + continue + } + liveJobs++ + for _, interval := range liveSplitJobIntervals(job, snapshot.Routes) { + if routeRangeIntersects(parent.Start, parent.End, interval.start, interval.end) { + return grpcStatusError(codes.Aborted, distribution.ErrSplitJobOverlap.Error()) + } + } + } + if liveJobs > 0 { + return grpcStatusError(codes.ResourceExhausted, distribution.ErrTooManyInFlightSplitJobs.Error()) + } + return nil +} + func splitJobIsLive(job distribution.SplitJob) bool { return job.Phase != distribution.SplitJobPhaseDone && job.Phase != distribution.SplitJobPhaseAbandoned } @@ -515,6 +883,201 @@ func routeRangeIntersects(aStart, aEnd, bStart, bEnd []byte) bool { return true } +func splitJobPassesListFilter(job distribution.SplitJob, sinceTerminalAtMs uint64, phaseFilter string) bool { + if sinceTerminalAtMs > 0 && job.TerminalAtMs > 0 && uint64(job.TerminalAtMs) < sinceTerminalAtMs { + return false + } + if phaseFilter == "" { + return true + } + return normalizeSplitJobPhaseFilter(pb.SplitJobPhase(job.Phase).String()) == phaseFilter || + normalizeSplitJobPhaseFilter(strings.TrimPrefix(pb.SplitJobPhase(job.Phase).String(), "SPLIT_JOB_PHASE_")) == phaseFilter +} + +func normalizeSplitJobPhaseFilter(phase string) string { + phase = strings.TrimSpace(phase) + if phase == "" { + return "" + } + phase = strings.ReplaceAll(phase, "-", "_") + phase = strings.ReplaceAll(phase, " ", "_") + return strings.ToUpper(phase) +} + +func validSplitJobPhaseFilter(phase string) bool { + for _, candidate := range pb.SplitJobPhase_name { + full := normalizeSplitJobPhaseFilter(candidate) + short := normalizeSplitJobPhaseFilter(strings.TrimPrefix(candidate, "SPLIT_JOB_PHASE_")) + if phase == full || phase == short { + return true + } + } + return false +} + +func splitJobCatalogStatusError(err error) error { + switch { + case errors.Is(err, distribution.ErrCatalogSplitJobIDRequired): + return grpcStatusError(codes.InvalidArgument, distribution.ErrCatalogSplitJobIDRequired.Error()) + case errors.Is(err, distribution.ErrCatalogSplitJobNotFound): + return grpcStatusError(codes.NotFound, distribution.ErrCatalogSplitJobNotFound.Error()) + case errors.Is(err, distribution.ErrCatalogSplitJobCannotRetry), + errors.Is(err, distribution.ErrCatalogSplitJobCannotAbandon): + return grpcStatusError(codes.FailedPrecondition, err.Error()) + case errors.Is(err, distribution.ErrCatalogSplitJobConflict): + return grpcStatusError(codes.Aborted, distribution.ErrCatalogSplitJobConflict.Error()) + default: + return grpcStatusErrorf(codes.Internal, "split job catalog: %v", err) + } +} + +func splitMigrationRangeStatusError(err error) error { + switch { + case errors.Is(err, distribution.ErrMigrationReservedRange), + errors.Is(err, distribution.ErrMigrationInvalidRoute): + return grpcStatusError(codes.InvalidArgument, err.Error()) + default: + return grpcStatusErrorf(codes.Internal, "validate split migration range: %v", err) + } +} + +func splitJobCoordinatorStatusError(err error) error { + switch { + case errors.Is(err, store.ErrWriteConflict): + return grpcStatusError(codes.Aborted, distribution.ErrCatalogSplitJobConflict.Error()) + case splitJobCoordinatorLeadershipError(err): + return grpcStatusError(codes.FailedPrecondition, errDistributionNotLeader.Error()) + default: + return grpcStatusErrorf(codes.Internal, "commit split job mutation: %v", err) + } +} + +func splitJobListPage(jobs []distribution.SplitJob, req *pb.ListSplitJobsRequest, phaseFilter string) (*pb.ListSplitJobsResponse, error) { + filtered := make([]distribution.SplitJob, 0, len(jobs)) + for _, job := range jobs { + if splitJobPassesListFilter(job, req.GetSinceTerminalAtMs(), phaseFilter) { + filtered = append(filtered, job) + } + } + sortSplitJobsForList(filtered) + + start, err := splitJobListStartIndex(filtered, req.GetPageCursor()) + if err != nil { + return nil, err + } + end := start + listSplitJobsDefaultPageSize + if end > len(filtered) { + end = len(filtered) + } + + resp := &pb.ListSplitJobsResponse{ + Jobs: make([]*pb.SplitJob, 0, end-start), + } + for _, job := range filtered[start:end] { + resp.Jobs = append(resp.Jobs, distribution.SplitJobToProto(job)) + } + if end < len(filtered) { + resp.NextPageCursor = encodeSplitJobListCursor(filtered[end-1]) + } + return resp, nil +} + +func sortSplitJobsForList(jobs []distribution.SplitJob) { + sort.Slice(jobs, func(i, j int) bool { + left, right := jobs[i], jobs[j] + leftLive := left.TerminalAtMs <= 0 + rightLive := right.TerminalAtMs <= 0 + if leftLive != rightLive { + return leftLive + } + if leftLive { + if left.UpdatedAtMs != right.UpdatedAtMs { + return left.UpdatedAtMs > right.UpdatedAtMs + } + return left.JobID > right.JobID + } + if left.TerminalAtMs != right.TerminalAtMs { + return left.TerminalAtMs > right.TerminalAtMs + } + return left.JobID > right.JobID + }) +} + +func splitJobListStartIndex(jobs []distribution.SplitJob, cursor []byte) (int, error) { + if len(cursor) == 0 { + return 0, nil + } + terminalAtMs, jobID, err := decodeSplitJobListCursor(cursor) + if err != nil { + return 0, err + } + for i, job := range jobs { + if job.JobID == jobID && splitJobListCursorTerminalAtMs(job) == terminalAtMs { + return i + 1, nil + } + } + return 0, grpcStatusError(codes.InvalidArgument, "split job page cursor does not match the filtered result set") +} + +func encodeSplitJobListCursor(job distribution.SplitJob) []byte { + cursor := make([]byte, splitJobListCursorEncodedBytes) + cursor[0] = splitJobListCursorVersion + binary.BigEndian.PutUint64( + cursor[splitJobListCursorTerminalOff:splitJobListCursorJobIDOff], + splitJobListCursorTerminalAtMs(job), + ) + binary.BigEndian.PutUint64(cursor[splitJobListCursorJobIDOff:], job.JobID) + return cursor +} + +func decodeSplitJobListCursor(cursor []byte) (uint64, uint64, error) { + if len(cursor) != splitJobListCursorEncodedBytes || cursor[0] != splitJobListCursorVersion { + return 0, 0, grpcStatusError(codes.InvalidArgument, "invalid split job page cursor") + } + terminalAtMs := binary.BigEndian.Uint64(cursor[splitJobListCursorTerminalOff:splitJobListCursorJobIDOff]) + jobID := binary.BigEndian.Uint64(cursor[splitJobListCursorJobIDOff:]) + if jobID == 0 { + return 0, 0, grpcStatusError(codes.InvalidArgument, "invalid split job page cursor") + } + return terminalAtMs, jobID, nil +} + +func splitJobListCursorTerminalAtMs(job distribution.SplitJob) uint64 { + if job.TerminalAtMs <= 0 { + return 0 + } + return uint64(job.TerminalAtMs) +} + +func splitJobCoordinatorLeadershipError(err error) bool { + if err == nil { + return false + } + if errors.Is(err, kv.ErrLeaderNotFound) || + errors.Is(err, raftengine.ErrNotLeader) || + errors.Is(err, raftengine.ErrLeadershipLost) || + errors.Is(err, raftengine.ErrLeadershipTransferInProgress) { + return true + } + return hasSplitJobLeaderErrorSuffix(err.Error()) +} + +var splitJobLeaderErrorPhrases = []string{ + "not leader", + "leader not found", + "leadership lost", + "leadership transfer in progress", +} + +func hasSplitJobLeaderErrorSuffix(msg string) bool { + for _, phrase := range splitJobLeaderErrorPhrases { + if len(msg) >= len(phrase) && strings.EqualFold(msg[len(msg)-len(phrase):], phrase) { + return true + } + } + return false +} + func splitCatalogRoutes( parent distribution.RouteDescriptor, splitKey []byte, @@ -573,13 +1136,13 @@ func (s *DistributionServer) allocateChildRouteIDs(ctx context.Context, readTS u return leftID, rightID, nil } -func findRouteByID(routes []distribution.RouteDescriptor, routeID uint64) (distribution.RouteDescriptor, int, bool) { - for i, route := range routes { +func findRouteByID(routes []distribution.RouteDescriptor, routeID uint64) (distribution.RouteDescriptor, bool) { + for _, route := range routes { if route.RouteID == routeID { - return distribution.CloneRouteDescriptor(route), i, true + return distribution.CloneRouteDescriptor(route), true } } - return distribution.RouteDescriptor{}, -1, false + return distribution.RouteDescriptor{}, false } func toProtoRouteDescriptors(routes []distribution.RouteDescriptor) []*pb.RouteDescriptor { diff --git a/adapter/distribution_server_test.go b/adapter/distribution_server_test.go index 7d4ba71f6..1f3fbaf04 100644 --- a/adapter/distribution_server_test.go +++ b/adapter/distribution_server_test.go @@ -3,10 +3,12 @@ package adapter import ( "bytes" "context" + "errors" "testing" "time" "github.com/bootjp/elastickv/distribution" + "github.com/bootjp/elastickv/internal/raftengine" "github.com/bootjp/elastickv/kv" pb "github.com/bootjp/elastickv/proto" "github.com/bootjp/elastickv/store" @@ -73,6 +75,24 @@ func TestWithCatalogReloadRetryPolicy_OverridesDefaults(t *testing.T) { require.Equal(t, 5*time.Millisecond, s.reloadRetry.interval) } +func TestDistributionServerPinReadTSWithoutTrackerReturnsReleasableToken(t *testing.T) { + t.Parallel() + + s := NewDistributionServer(distribution.NewEngine(), nil) + token := s.pinReadTS(10) + require.NotNil(t, token) + require.NotPanics(t, func() { + token.Release() + }) + + var nilServer *DistributionServer + nilToken := nilServer.pinReadTS(10) + require.NotNil(t, nilToken) + require.NotPanics(t, func() { + nilToken.Release() + }) +} + func TestDistributionServerListRoutes_ReadsDurableCatalog(t *testing.T) { t.Parallel() @@ -130,6 +150,602 @@ func TestDistributionServerListRoutes_RequiresCatalog(t *testing.T) { require.ErrorContains(t, err, errDistributionCatalogNotConfigured.Error()) } +func TestDistributionServerGetSplitMigrationCapabilityReportsNotReadyUntilRunnerReady(t *testing.T) { + t.Parallel() + + s := NewDistributionServer(distribution.NewEngine(), nil) + resp, err := s.GetSplitMigrationCapability(context.Background(), &pb.GetSplitMigrationCapabilityRequest{}) + require.NoError(t, err) + require.False(t, resp.GetMigrationCapable()) + require.NotContains(t, resp.GetCapabilities(), splitMigrationCapabilityV2) +} + +func TestDistributionServerGetSplitMigrationCapabilityReportsReadyWhenRunnerReady(t *testing.T) { + t.Parallel() + + s := NewDistributionServer(distribution.NewEngine(), nil, WithSplitJobRunnerReady()) + resp, err := s.GetSplitMigrationCapability(context.Background(), &pb.GetSplitMigrationCapabilityRequest{}) + require.NoError(t, err) + require.True(t, resp.GetMigrationCapable()) + require.Contains(t, resp.GetCapabilities(), splitMigrationCapabilityV2) +} + +func TestDistributionServerStartSplitMigration_FailsClosedUntilCapabilityGate(t *testing.T) { + t.Parallel() + + ctx := context.Background() + baseStore := store.NewMVCCStore() + catalog := distribution.NewCatalogStore(baseStore) + coordinator := newDistributionCoordinatorStub(baseStore, true) + s := NewDistributionServer(distribution.NewEngine(), catalog, WithDistributionCoordinator(coordinator)) + + _, err := s.StartSplitMigration(ctx, &pb.StartSplitMigrationRequest{ + ExpectedCatalogVersion: 1, + RouteId: 1, + SplitKey: []byte("g"), + TargetGroupId: 2, + }) + require.Error(t, err) + require.Equal(t, codes.FailedPrecondition, status.Code(err)) + require.ErrorContains(t, err, errDistributionClusterNotReady.Error()) + require.Zero(t, coordinator.dispatchCalls) + + jobs, listErr := catalog.ListSplitJobs(ctx) + require.NoError(t, listErr) + require.Empty(t, jobs) +} + +func TestDistributionServerStartSplitMigrationReturnsCapabilityGateError(t *testing.T) { + t.Parallel() + + ctx := context.Background() + baseStore := store.NewMVCCStore() + catalog := distribution.NewCatalogStore(baseStore) + coordinator := newDistributionCoordinatorStub(baseStore, true) + s := NewDistributionServer( + distribution.NewEngine(), + catalog, + WithDistributionCoordinator(coordinator), + WithSplitMigrationCapabilityGate(func(context.Context) error { + return status.Error(codes.Unavailable, "split migration capability not ready") + }), + ) + + _, err := s.StartSplitMigration(ctx, &pb.StartSplitMigrationRequest{ + ExpectedCatalogVersion: 1, + RouteId: 1, + SplitKey: []byte("g"), + TargetGroupId: 2, + }) + require.Error(t, err) + require.Equal(t, codes.Unavailable, status.Code(err)) + require.ErrorContains(t, err, "split migration capability not ready") + require.Zero(t, coordinator.dispatchCalls) +} + +func TestDistributionServerStartSplitMigrationCapabilityGateRunsOutsideCatalogLock(t *testing.T) { + ctx := context.Background() + baseStore := store.NewMVCCStore() + catalog := distribution.NewCatalogStore(baseStore) + saved, err := catalog.Save(ctx, 0, []distribution.RouteDescriptor{{ + RouteID: 1, + Start: []byte("a"), + End: []byte("m"), + GroupID: 1, + State: distribution.RouteStateActive, + ParentRouteID: 0, + }}) + require.NoError(t, err) + + coordinator := newDistributionCoordinatorStub(baseStore, true) + gateEntered := make(chan struct{}) + releaseGate := make(chan struct{}) + s := NewDistributionServer( + distribution.NewEngine(), + catalog, + WithDistributionCoordinator(coordinator), + WithDistributionKnownRaftGroups(1, 2), + WithSplitMigrationCapabilityGate(func(context.Context) error { + close(gateEntered) + <-releaseGate + return status.Error(codes.Unavailable, "split migration capability not ready") + }), + ) + + startDone := make(chan error, 1) + go func() { + _, err := s.StartSplitMigration(ctx, &pb.StartSplitMigrationRequest{ + ExpectedCatalogVersion: saved.Version, + RouteId: 1, + SplitKey: []byte("g"), + TargetGroupId: 2, + }) + startDone <- err + }() + <-gateEntered + + splitDone := make(chan error, 1) + go func() { + _, err := s.SplitRange(ctx, &pb.SplitRangeRequest{ + ExpectedCatalogVersion: saved.Version, + RouteId: 1, + SplitKey: []byte("g"), + }) + splitDone <- err + }() + select { + case err := <-splitDone: + require.NoError(t, err) + case <-time.After(500 * time.Millisecond): + t.Fatal("SplitRange blocked behind StartSplitMigration capability gate") + } + + close(releaseGate) + err = <-startDone + require.Error(t, err) + require.Equal(t, codes.Unavailable, status.Code(err)) +} + +func TestDistributionServerStartSplitMigration_CreatesPlannedJobWhenGateOpen(t *testing.T) { + t.Parallel() + + ctx := context.Background() + baseStore := store.NewMVCCStore() + catalog := distribution.NewCatalogStore(baseStore) + saved, err := catalog.Save(ctx, 0, []distribution.RouteDescriptor{{ + RouteID: 1, + Start: []byte("a"), + End: []byte("m"), + GroupID: 1, + State: distribution.RouteStateActive, + ParentRouteID: 0, + }}) + require.NoError(t, err) + + coordinator := newDistributionCoordinatorStub(baseStore, true) + gateCalls := 0 + s := NewDistributionServer( + distribution.NewEngine(), + catalog, + WithDistributionCoordinator(coordinator), + WithDistributionKnownRaftGroups(1, 2), + WithSplitMigrationCapabilityGate(func(context.Context) error { + gateCalls++ + return nil + }), + ) + + resp, err := s.StartSplitMigration(ctx, &pb.StartSplitMigrationRequest{ + ExpectedCatalogVersion: saved.Version, + RouteId: 1, + SplitKey: []byte("g"), + TargetGroupId: 2, + }) + require.NoError(t, err) + require.Equal(t, saved.Version, resp.CatalogVersion) + require.Equal(t, uint64(1), resp.JobId) + require.Equal(t, 1, gateCalls) + require.Equal(t, 1, coordinator.dispatchCalls) + require.ElementsMatch(t, []string{ + string(distribution.CatalogSplitJobKey(1)), + string(distribution.CatalogNextSplitJobIDKey()), + string(distribution.CatalogVersionKey()), + string(distribution.CatalogRouteKey(1)), + }, byteSliceStrings(coordinator.lastReadKeys)) + + jobs, err := catalog.ListSplitJobs(ctx) + require.NoError(t, err) + require.Len(t, jobs, 1) + job := jobs[0] + require.Equal(t, uint64(1), job.JobID) + require.Equal(t, uint64(1), job.SourceRouteID) + require.Equal(t, []byte("g"), job.SplitKey) + require.Equal(t, uint64(2), job.TargetGroupID) + require.Equal(t, distribution.SplitJobPhasePlanned, job.Phase) + require.NotZero(t, job.StartedAtMs) + require.NotZero(t, job.UpdatedAtMs) + require.NotEmpty(t, job.BracketProgress) + next, err := catalog.NextSplitJobID(ctx) + require.NoError(t, err) + require.Equal(t, uint64(2), next) +} + +func TestDistributionServerStartSplitMigration_RejectsUnknownTargetGroup(t *testing.T) { + t.Parallel() + + ctx := context.Background() + baseStore := store.NewMVCCStore() + catalog := distribution.NewCatalogStore(baseStore) + saved, err := catalog.Save(ctx, 0, []distribution.RouteDescriptor{{ + RouteID: 1, + Start: []byte("a"), + End: []byte("m"), + GroupID: 1, + State: distribution.RouteStateActive, + ParentRouteID: 0, + }}) + require.NoError(t, err) + + coordinator := newDistributionCoordinatorStub(baseStore, true) + s := NewDistributionServer( + distribution.NewEngine(), + catalog, + WithDistributionCoordinator(coordinator), + WithDistributionKnownRaftGroups(1, 2), + WithSplitMigrationCapabilityGate(func(context.Context) error { return nil }), + ) + + _, err = s.StartSplitMigration(ctx, &pb.StartSplitMigrationRequest{ + ExpectedCatalogVersion: saved.Version, + RouteId: 1, + SplitKey: []byte("g"), + TargetGroupId: 3, + }) + require.Error(t, err) + require.Equal(t, codes.InvalidArgument, status.Code(err)) + require.ErrorContains(t, err, errDistributionUnknownTargetGroup.Error()) + require.Zero(t, coordinator.dispatchCalls) +} + +func TestDistributionServerStartSplitMigration_RejectsNonActiveSourceRoute(t *testing.T) { + t.Parallel() + + ctx := context.Background() + for _, tc := range []struct { + name string + state distribution.RouteState + }{ + {name: "write_fenced", state: distribution.RouteStateWriteFenced}, + {name: "migrating_source", state: distribution.RouteStateMigratingSource}, + {name: "migrating_target", state: distribution.RouteStateMigratingTarget}, + } { + t.Run(tc.name, func(t *testing.T) { + baseStore := store.NewMVCCStore() + catalog := distribution.NewCatalogStore(baseStore) + saved, err := catalog.Save(ctx, 0, []distribution.RouteDescriptor{{ + RouteID: 1, + Start: []byte("a"), + End: []byte("m"), + GroupID: 1, + State: tc.state, + ParentRouteID: 0, + }}) + require.NoError(t, err) + + coordinator := newDistributionCoordinatorStub(baseStore, true) + s := NewDistributionServer( + distribution.NewEngine(), + catalog, + WithDistributionCoordinator(coordinator), + WithDistributionKnownRaftGroups(1, 2), + WithSplitMigrationCapabilityGate(func(context.Context) error { return nil }), + ) + + _, err = s.StartSplitMigration(ctx, &pb.StartSplitMigrationRequest{ + ExpectedCatalogVersion: saved.Version, + RouteId: 1, + SplitKey: []byte("g"), + TargetGroupId: 2, + }) + require.Error(t, err) + require.Equal(t, codes.FailedPrecondition, status.Code(err)) + require.ErrorContains(t, err, errDistributionSourceRouteNotActive.Error()) + require.Zero(t, coordinator.dispatchCalls) + + jobs, listErr := catalog.ListSplitJobs(ctx) + require.NoError(t, listErr) + require.Empty(t, jobs) + }) + } +} + +func TestDistributionServerStartSplitMigration_RejectsSecondLiveJob(t *testing.T) { + t.Parallel() + + ctx := context.Background() + baseStore := store.NewMVCCStore() + catalog := distribution.NewCatalogStore(baseStore) + saved, err := catalog.Save(ctx, 0, []distribution.RouteDescriptor{ + { + RouteID: 1, + Start: []byte("a"), + End: []byte("m"), + GroupID: 1, + State: distribution.RouteStateActive, + ParentRouteID: 0, + }, + { + RouteID: 2, + Start: []byte("m"), + End: nil, + GroupID: 2, + State: distribution.RouteStateActive, + ParentRouteID: 0, + }, + }) + require.NoError(t, err) + require.NoError(t, catalog.CreateSplitJob(ctx, distribution.SplitJob{ + JobID: 1, + SourceRouteID: 2, + SplitKey: []byte("t"), + TargetGroupID: 3, + Phase: distribution.SplitJobPhaseBackfill, + StartedAtMs: 1000, + UpdatedAtMs: 1000, + })) + + coordinator := newDistributionCoordinatorStub(baseStore, true) + s := NewDistributionServer( + distribution.NewEngine(), + catalog, + WithDistributionCoordinator(coordinator), + WithDistributionKnownRaftGroups(1, 2, 3, 4), + WithSplitMigrationCapabilityGate(func(context.Context) error { return nil }), + ) + + _, err = s.StartSplitMigration(ctx, &pb.StartSplitMigrationRequest{ + ExpectedCatalogVersion: saved.Version, + RouteId: 1, + SplitKey: []byte("g"), + TargetGroupId: 4, + }) + require.Error(t, err) + require.Equal(t, codes.ResourceExhausted, status.Code(err)) + require.ErrorContains(t, err, distribution.ErrTooManyInFlightSplitJobs.Error()) + require.Zero(t, coordinator.dispatchCalls) +} + +func TestDistributionServerStartSplitMigration_RejectsReservedMigrationRange(t *testing.T) { + t.Parallel() + + ctx := context.Background() + for _, tc := range []struct { + name string + splitKey []byte + end []byte + }{ + {name: "dist catalog", splitKey: []byte("!dist|"), end: nil}, + {name: "dist staged catalog", splitKey: []byte("!dist|migstage|"), end: nil}, + {name: "migration staged", splitKey: []byte("!migstage|ready|1"), end: nil}, + {name: "migration tracker", splitKey: []byte("!migwrite|"), end: nil}, + {name: "migration fence", splitKey: []byte("!migfence|"), end: nil}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + baseStore := store.NewMVCCStore() + catalog := distribution.NewCatalogStore(baseStore) + saved, err := catalog.Save(ctx, 0, []distribution.RouteDescriptor{{ + RouteID: 1, + Start: []byte(""), + End: tc.end, + GroupID: 1, + State: distribution.RouteStateActive, + ParentRouteID: 0, + }}) + require.NoError(t, err) + + coordinator := newDistributionCoordinatorStub(baseStore, true) + s := NewDistributionServer( + distribution.NewEngine(), + catalog, + WithDistributionCoordinator(coordinator), + WithDistributionKnownRaftGroups(1, 2), + WithSplitMigrationCapabilityGate(func(context.Context) error { return nil }), + ) + + _, err = s.StartSplitMigration(ctx, &pb.StartSplitMigrationRequest{ + ExpectedCatalogVersion: saved.Version, + RouteId: 1, + SplitKey: tc.splitKey, + TargetGroupId: 2, + }) + require.Error(t, err) + require.Equal(t, codes.InvalidArgument, status.Code(err)) + require.ErrorContains(t, err, distribution.ErrMigrationReservedRange.Error()) + require.Zero(t, coordinator.dispatchCalls) + + jobs, listErr := catalog.ListSplitJobs(ctx) + require.NoError(t, listErr) + require.Empty(t, jobs) + }) + } +} + +func TestDistributionServerSplitJobRPCs_ReadAndListCatalogJobs(t *testing.T) { + t.Parallel() + + ctx := context.Background() + catalog := distribution.NewCatalogStore(store.NewMVCCStore()) + live := sampleDistributionSplitJob(10) + live.Phase = distribution.SplitJobPhaseDeltaCopy + require.NoError(t, catalog.CreateSplitJob(ctx, live)) + history := sampleDistributionSplitJob(11) + history.Phase = distribution.SplitJobPhaseDone + history.TerminalAtMs = 2000 + require.NoError(t, catalog.CreateSplitJob(ctx, history)) + require.NoError(t, catalog.MoveSplitJobToHistory(ctx, history, history)) + + s := NewDistributionServer(distribution.NewEngine(), catalog) + got, err := s.GetSplitJob(ctx, &pb.GetSplitJobRequest{JobId: live.JobID}) + require.NoError(t, err) + require.Equal(t, live.JobID, got.Job.JobId) + require.Equal(t, pb.SplitJobPhase_SPLIT_JOB_PHASE_DELTA_COPY, got.Job.Phase) + + listAll, err := s.ListSplitJobs(ctx, &pb.ListSplitJobsRequest{}) + require.NoError(t, err) + require.Len(t, listAll.Jobs, 2) + require.Equal(t, []uint64{live.JobID, history.JobID}, []uint64{listAll.Jobs[0].JobId, listAll.Jobs[1].JobId}) + + listDone, err := s.ListSplitJobs(ctx, &pb.ListSplitJobsRequest{Phase: "done"}) + require.NoError(t, err) + require.Len(t, listDone.Jobs, 1) + require.Equal(t, history.JobID, listDone.Jobs[0].JobId) + + _, err = s.ListSplitJobs(ctx, &pb.ListSplitJobsRequest{Phase: "not-a-phase"}) + require.Error(t, err) + require.Equal(t, codes.InvalidArgument, status.Code(err)) +} + +func TestDistributionServerListSplitJobs_PaginatesNewestHistory(t *testing.T) { + t.Parallel() + + ctx := context.Background() + catalog := distribution.NewCatalogStore(store.NewMVCCStore()) + for jobID, terminalAtMs := uint64(1), int64(1001); jobID <= 205; jobID, terminalAtMs = jobID+1, terminalAtMs+1 { + job := sampleDistributionSplitJob(jobID) + job.Phase = distribution.SplitJobPhaseDone + job.TerminalAtMs = terminalAtMs + job.UpdatedAtMs = job.TerminalAtMs + require.NoError(t, catalog.CreateSplitJob(ctx, job)) + require.NoError(t, catalog.MoveSplitJobToHistory(ctx, job, job)) + } + + s := NewDistributionServer(distribution.NewEngine(), catalog) + first, err := s.ListSplitJobs(ctx, &pb.ListSplitJobsRequest{}) + require.NoError(t, err) + require.Len(t, first.Jobs, listSplitJobsDefaultPageSize) + require.NotEmpty(t, first.NextPageCursor) + require.Equal(t, uint64(205), first.Jobs[0].JobId) + require.Equal(t, uint64(6), first.Jobs[len(first.Jobs)-1].JobId) + + second, err := s.ListSplitJobs(ctx, &pb.ListSplitJobsRequest{PageCursor: first.NextPageCursor}) + require.NoError(t, err) + require.Empty(t, second.NextPageCursor) + require.Equal(t, []uint64{5, 4, 3, 2, 1}, splitJobIDs(second.Jobs)) +} + +func TestDistributionServerListSplitJobs_RejectsInvalidCursor(t *testing.T) { + t.Parallel() + + ctx := context.Background() + catalog := distribution.NewCatalogStore(store.NewMVCCStore()) + job := sampleDistributionSplitJob(1) + job.Phase = distribution.SplitJobPhaseDone + job.TerminalAtMs = 1000 + require.NoError(t, catalog.CreateSplitJob(ctx, job)) + require.NoError(t, catalog.MoveSplitJobToHistory(ctx, job, job)) + + s := NewDistributionServer(distribution.NewEngine(), catalog) + _, err := s.ListSplitJobs(ctx, &pb.ListSplitJobsRequest{PageCursor: []byte("bad")}) + require.Error(t, err) + require.Equal(t, codes.InvalidArgument, status.Code(err)) + + missing := sampleDistributionSplitJob(999) + missing.TerminalAtMs = 9999 + _, err = s.ListSplitJobs(ctx, &pb.ListSplitJobsRequest{PageCursor: encodeSplitJobListCursor(missing)}) + require.Error(t, err) + require.Equal(t, codes.InvalidArgument, status.Code(err)) +} + +func TestDistributionServerRetrySplitJob_UsesCoordinatorCAS(t *testing.T) { + t.Parallel() + + ctx := context.Background() + baseStore := store.NewMVCCStore() + catalog := distribution.NewCatalogStore(baseStore) + job := sampleDistributionSplitJob(12) + job.Phase = distribution.SplitJobPhaseFailed + job.RetryPhase = distribution.SplitJobPhaseFence + job.LastError = "retry me" + require.NoError(t, catalog.CreateSplitJob(ctx, job)) + + coordinator := newDistributionCoordinatorStub(baseStore, true) + s := NewDistributionServer(distribution.NewEngine(), catalog, WithDistributionCoordinator(coordinator)) + _, err := s.RetrySplitJob(ctx, &pb.RetrySplitJobRequest{JobId: job.JobID}) + require.NoError(t, err) + require.Equal(t, 1, coordinator.dispatchCalls) + + got, found, err := catalog.SplitJob(ctx, job.JobID) + require.NoError(t, err) + require.True(t, found) + require.Equal(t, distribution.SplitJobPhaseFence, got.Phase) + require.Equal(t, distribution.SplitJobPhaseNone, got.RetryPhase) + require.Empty(t, got.LastError) +} + +func TestDistributionServerRetrySplitJob_MapsDispatchLeadershipLoss(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + err error + }{ + {name: "leader not found", err: kv.ErrLeaderNotFound}, + {name: "raft not leader", err: raftengine.ErrNotLeader}, + {name: "leadership lost", err: raftengine.ErrLeadershipLost}, + {name: "transfer in progress", err: raftengine.ErrLeadershipTransferInProgress}, + {name: "wrapped grpc detail", err: errors.New("rpc error: code = Unknown desc = raft engine: leadership lost")}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + ctx := context.Background() + baseStore := store.NewMVCCStore() + catalog := distribution.NewCatalogStore(baseStore) + job := sampleDistributionSplitJob(15) + job.Phase = distribution.SplitJobPhaseFailed + job.RetryPhase = distribution.SplitJobPhaseFence + require.NoError(t, catalog.CreateSplitJob(ctx, job)) + + coordinator := newDistributionCoordinatorStub(baseStore, true) + coordinator.dispatchErr = tc.err + s := NewDistributionServer(distribution.NewEngine(), catalog, WithDistributionCoordinator(coordinator)) + _, err := s.RetrySplitJob(ctx, &pb.RetrySplitJobRequest{JobId: job.JobID}) + require.Error(t, err) + require.Equal(t, codes.FailedPrecondition, status.Code(err)) + require.ErrorContains(t, err, errDistributionNotLeader.Error()) + require.Equal(t, 1, coordinator.dispatchCalls) + }) + } +} + +func TestDistributionServerAbandonSplitJob_RecordsAbandoningViaCoordinator(t *testing.T) { + t.Parallel() + + ctx := context.Background() + baseStore := store.NewMVCCStore() + catalog := distribution.NewCatalogStore(baseStore) + job := sampleDistributionSplitJob(13) + job.Phase = distribution.SplitJobPhaseFailed + job.RetryPhase = distribution.SplitJobPhaseBackfill + job.AbandonFromPhase = distribution.SplitJobPhaseNone + require.NoError(t, catalog.CreateSplitJob(ctx, job)) + + coordinator := newDistributionCoordinatorStub(baseStore, true) + s := NewDistributionServer(distribution.NewEngine(), catalog, WithDistributionCoordinator(coordinator)) + _, err := s.AbandonSplitJob(ctx, &pb.AbandonSplitJobRequest{JobId: job.JobID}) + require.NoError(t, err) + require.Equal(t, 1, coordinator.dispatchCalls) + + got, found, err := catalog.SplitJob(ctx, job.JobID) + require.NoError(t, err) + require.True(t, found) + require.Equal(t, distribution.SplitJobPhaseAbandoning, got.Phase) + require.Equal(t, distribution.SplitJobPhaseNone, got.RetryPhase) + require.Equal(t, distribution.SplitJobPhaseBackfill, got.AbandonFromPhase) +} + +func TestDistributionServerRetrySplitJob_RequiresCatalogLeader(t *testing.T) { + t.Parallel() + + ctx := context.Background() + baseStore := store.NewMVCCStore() + catalog := distribution.NewCatalogStore(baseStore) + job := sampleDistributionSplitJob(14) + job.Phase = distribution.SplitJobPhaseFailed + job.RetryPhase = distribution.SplitJobPhaseBackfill + require.NoError(t, catalog.CreateSplitJob(ctx, job)) + + coordinator := newDistributionCoordinatorStub(baseStore, false) + s := NewDistributionServer(distribution.NewEngine(), catalog, WithDistributionCoordinator(coordinator)) + _, err := s.RetrySplitJob(ctx, &pb.RetrySplitJobRequest{JobId: job.JobID}) + require.Error(t, err) + require.Equal(t, codes.FailedPrecondition, status.Code(err)) + require.Zero(t, coordinator.dispatchCalls) +} + func TestDistributionServerGetRouteOwnership_UsesExactVersionSnapshot(t *testing.T) { t.Parallel() @@ -830,9 +1446,39 @@ func seededDistributionServerWithoutCoordinator(t *testing.T) (*DistributionServ return NewDistributionServer(distribution.NewEngine(), catalog), saved.Version } +func sampleDistributionSplitJob(jobID uint64) distribution.SplitJob { + return distribution.SplitJob{ + JobID: jobID, + SourceRouteID: 1, + SplitKey: []byte("g"), + TargetGroupID: 2, + Phase: distribution.SplitJobPhaseBackfill, + RetryPhase: distribution.SplitJobPhaseNone, + StartedAtMs: 1000, + UpdatedAtMs: 1000, + } +} + +func splitJobIDs(jobs []*pb.SplitJob) []uint64 { + ids := make([]uint64, 0, len(jobs)) + for _, job := range jobs { + ids = append(ids, job.GetJobId()) + } + return ids +} + +func byteSliceStrings(in [][]byte) []string { + out := make([]string, 0, len(in)) + for _, item := range in { + out = append(out, string(item)) + } + return out +} + type distributionCoordinatorStub struct { store store.MVCCStore leader bool + dispatchErr error nextTS uint64 lastStartTS uint64 lastReadKeys [][]byte @@ -855,6 +1501,9 @@ func (s *distributionCoordinatorStub) Dispatch(ctx context.Context, reqs *kv.Ope return nil, err } s.dispatchCalls++ + if s.dispatchErr != nil { + return nil, s.dispatchErr + } startTS, commitTS := s.nextTimestamps(reqs.StartTS) s.lastStartTS = startTS readKeys := cloneDistributionReadKeys(reqs.ReadKeys) diff --git a/adapter/redis_delta_compactor_test.go b/adapter/redis_delta_compactor_test.go index 12ab9513a..7892e0b37 100644 --- a/adapter/redis_delta_compactor_test.go +++ b/adapter/redis_delta_compactor_test.go @@ -499,27 +499,20 @@ 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) - - runCtx, cancel := context.WithCancel(ctx) - defer cancel() - go func() { _ = c.Run(runCtx) }() + // 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}) - // Wait until the base meta holds the accumulated total. - // The pagination loop should take two passes: first 1025, then 9. - require.Eventually(t, func() bool { - readTS := st.LastCommitTS() - raw, err := st.GetAt(ctx, store.HashMetaKey(userKey), readTS) - if err != nil { - return false - } - got, err := store.UnmarshalHashMeta(raw) - return err == nil && got.Len == int64(totalDeltasU64) - }, 5*time.Second, 20*time.Millisecond, "all %d delta keys should be compacted into base meta", totalDeltasU64) + readTS := st.LastCommitTS() + raw, err := st.GetAt(ctx, store.HashMetaKey(userKey), readTS) + require.NoError(t, err) + got, err := store.UnmarshalHashMeta(raw) + require.NoError(t, err) + require.Equal(t, int64(totalDeltasU64), got.Len, "all %d delta keys should be compacted into base meta", totalDeltasU64) // No delta keys should remain after pagination compaction. - readTS := st.LastCommitTS() prefix := store.HashMetaDeltaScanPrefix(userKey) end := store.PrefixScanEnd(prefix) remaining, err := st.ScanAt(ctx, prefix, end, int(totalDeltasU64)+1, readTS) diff --git a/distribution/split_job_catalog.go b/distribution/split_job_catalog.go index 808ad8498..1f7fc241b 100644 --- a/distribution/split_job_catalog.go +++ b/distribution/split_job_catalog.go @@ -36,6 +36,7 @@ var ( ErrCatalogSplitJobConflict = errors.New("catalog split job conflict") ErrCatalogSplitJobTerminalRequired = errors.New("catalog split job terminal state is required") ErrSplitJobOverlap = errors.New("split job overlaps requested route") + ErrTooManyInFlightSplitJobs = errors.New("too many in-flight split jobs") ) // SplitJobPhase is the durable phase of a split migration job. @@ -844,6 +845,11 @@ func CloneSplitJob(job SplitJob) SplitJob { } } +// SplitJobToProto converts a catalog SplitJob into its wire representation. +func SplitJobToProto(job SplitJob) *pb.SplitJob { + return splitJobToProto(job) +} + func splitJobToProto(job SplitJob) *pb.SplitJob { job = CloneSplitJob(job) return &pb.SplitJob{ diff --git a/distribution/split_job_catalog_test.go b/distribution/split_job_catalog_test.go index 0ac91a59b..cdeef3c3a 100644 --- a/distribution/split_job_catalog_test.go +++ b/distribution/split_job_catalog_test.go @@ -242,6 +242,98 @@ func TestCatalogStoreSaveSplitJobRejectsStaleExpectedJob(t *testing.T) { assertSplitJobEqual(t, advanced, got) } +func TestCatalogStoreRetrySplitJobUsesDurableRetryPhase(t *testing.T) { + cs := NewCatalogStore(store.NewMVCCStore()) + ctx := context.Background() + job := sampleSplitJob(18) + job.Phase = SplitJobPhaseFailed + job.RetryPhase = SplitJobPhaseDeltaCopy + job.LastError = "transient" + + if err := cs.CreateSplitJob(ctx, job); err != nil { + t.Fatalf("create split job: %v", err) + } + retried, err := cs.RetrySplitJob(ctx, job.JobID, 1200) + if err != nil { + t.Fatalf("retry split job: %v", err) + } + if retried.Phase != SplitJobPhaseDeltaCopy || retried.RetryPhase != SplitJobPhaseNone { + t.Fatalf("unexpected retry transition: %+v", retried) + } + if retried.AbandonFromPhase != SplitJobPhaseNone || retried.LastError != "" || retried.UpdatedAtMs != 1200 { + t.Fatalf("unexpected retry witness cleanup: %+v", retried) + } + got, found, err := cs.SplitJob(ctx, job.JobID) + if err != nil { + t.Fatalf("load retried split job: %v", err) + } + if !found { + t.Fatal("expected retried split job") + } + assertSplitJobEqual(t, retried, got) +} + +func TestCatalogStoreRetrySplitJobRejectsMissingRetryWitness(t *testing.T) { + cs := NewCatalogStore(store.NewMVCCStore()) + ctx := context.Background() + job := sampleSplitJob(19) + job.Phase = SplitJobPhaseFailed + job.RetryPhase = SplitJobPhaseNone + + if err := cs.CreateSplitJob(ctx, job); err != nil { + t.Fatalf("create split job: %v", err) + } + if _, err := cs.RetrySplitJob(ctx, job.JobID, 1200); !errors.Is(err, ErrCatalogSplitJobCannotRetry) { + t.Fatalf("expected ErrCatalogSplitJobCannotRetry, got %v", err) + } +} + +func TestCatalogStoreBeginSplitJobAbandonRecordsPreCutoverPhase(t *testing.T) { + cs := NewCatalogStore(store.NewMVCCStore()) + ctx := context.Background() + job := sampleSplitJob(20) + job.Phase = SplitJobPhaseFailed + job.RetryPhase = SplitJobPhaseFence + job.AbandonFromPhase = SplitJobPhaseNone + + if err := cs.CreateSplitJob(ctx, job); err != nil { + t.Fatalf("create split job: %v", err) + } + abandoning, err := cs.BeginSplitJobAbandon(ctx, job.JobID, 1300) + if err != nil { + t.Fatalf("begin split job abandon: %v", err) + } + if abandoning.Phase != SplitJobPhaseAbandoning || + abandoning.RetryPhase != SplitJobPhaseNone || + abandoning.AbandonFromPhase != SplitJobPhaseFence || + abandoning.UpdatedAtMs != 1300 { + t.Fatalf("unexpected abandon transition: %+v", abandoning) + } + got, found, err := cs.SplitJob(ctx, job.JobID) + if err != nil { + t.Fatalf("load abandoning split job: %v", err) + } + if !found { + t.Fatal("expected abandoning split job") + } + assertSplitJobEqual(t, abandoning, got) +} + +func TestCatalogStoreBeginSplitJobAbandonRejectsPostCutoverPhases(t *testing.T) { + cs := NewCatalogStore(store.NewMVCCStore()) + ctx := context.Background() + job := sampleSplitJob(21) + job.Phase = SplitJobPhaseCleanup + job.RetryPhase = SplitJobPhaseNone + + if err := cs.CreateSplitJob(ctx, job); err != nil { + t.Fatalf("create split job: %v", err) + } + if _, err := cs.BeginSplitJobAbandon(ctx, job.JobID, 1300); !errors.Is(err, ErrCatalogSplitJobCannotAbandon) { + t.Fatalf("expected ErrCatalogSplitJobCannotAbandon, got %v", err) + } +} + func TestCatalogStoreListSplitJobsIncludesLiveAndHistory(t *testing.T) { cs := NewCatalogStore(store.NewMVCCStore()) ctx := context.Background() diff --git a/distribution/split_job_lifecycle.go b/distribution/split_job_lifecycle.go new file mode 100644 index 000000000..f2427a4e0 --- /dev/null +++ b/distribution/split_job_lifecycle.go @@ -0,0 +1,147 @@ +package distribution + +import ( + "bytes" + "context" + + "github.com/cockroachdb/errors" +) + +var ( + ErrCatalogSplitJobNotFound = errors.New("catalog split job is not found") + ErrCatalogSplitJobCannotRetry = errors.New("catalog split job cannot be retried") + ErrCatalogSplitJobCannotAbandon = errors.New("catalog split job cannot be abandoned") +) + +// RetrySplitJobState returns the durable state transition for RetrySplitJob. +// The retry target must come from the recorded RetryPhase witness; callers must +// not infer a phase from cursors, timestamps, or diagnostics after restart. +func RetrySplitJobState(job SplitJob, nowMs int64) (SplitJob, error) { + out := CloneSplitJob(job) + if out.Phase != SplitJobPhaseFailed || !splitJobRetryPhaseAllowed(out.RetryPhase) { + return SplitJob{}, errors.WithStack(ErrCatalogSplitJobCannotRetry) + } + out.Phase = out.RetryPhase + out.RetryPhase = SplitJobPhaseNone + out.AbandonFromPhase = SplitJobPhaseNone + out.LastError = "" + out.UpdatedAtMs = nowMs + return out, nil +} + +// BeginSplitJobAbandon records the durable ABANDONING witness before any +// pre-CUTOVER cleanup side effect is removed. +func BeginSplitJobAbandon(job SplitJob, nowMs int64) (SplitJob, error) { + out := CloneSplitJob(job) + from, ok := splitJobAbandonFromPhase(out) + if !ok { + return SplitJob{}, errors.WithStack(ErrCatalogSplitJobCannotAbandon) + } + if out.Phase == SplitJobPhaseAbandoning { + return out, nil + } + out.Phase = SplitJobPhaseAbandoning + out.RetryPhase = SplitJobPhaseNone + out.AbandonFromPhase = from + out.UpdatedAtMs = nowMs + return out, nil +} + +// RetrySplitJob CASes a FAILED live job back to its recorded retry phase. +func (s *CatalogStore) RetrySplitJob(ctx context.Context, jobID uint64, nowMs int64) (SplitJob, error) { + expected, _, err := s.LiveSplitJobForUpdate(ctx, jobID) + if err != nil { + return SplitJob{}, err + } + next, err := RetrySplitJobState(expected, nowMs) + if err != nil { + return SplitJob{}, err + } + return next, s.SaveSplitJob(ctx, expected, next) +} + +// BeginSplitJobAbandon CASes a live pre-CUTOVER job into ABANDONING. +func (s *CatalogStore) BeginSplitJobAbandon(ctx context.Context, jobID uint64, nowMs int64) (SplitJob, error) { + expected, _, err := s.LiveSplitJobForUpdate(ctx, jobID) + if err != nil { + return SplitJob{}, err + } + next, err := BeginSplitJobAbandon(expected, nowMs) + if err != nil { + return SplitJob{}, err + } + if SplitJobsEquivalent(expected, next) { + return next, nil + } + return next, s.SaveSplitJob(ctx, expected, next) +} + +// LiveSplitJobForUpdate reads the latest live split job and the snapshot +// timestamp used for the read. RPC callers use readTS as a CAS StartTS when +// proposing the update through Raft. +func (s *CatalogStore) LiveSplitJobForUpdate(ctx context.Context, jobID uint64) (SplitJob, uint64, error) { + if err := ensureCatalogStore(s); err != nil { + return SplitJob{}, 0, err + } + if jobID == 0 { + return SplitJob{}, 0, errors.WithStack(ErrCatalogSplitJobIDRequired) + } + ctx = contextOrBackground(ctx) + readTS := s.store.LastCommitTS() + job, found, err := s.liveSplitJobAt(ctx, jobID, readTS) + if err != nil { + return SplitJob{}, 0, err + } + if !found { + return SplitJob{}, 0, errors.WithStack(ErrCatalogSplitJobNotFound) + } + return job, readTS, nil +} + +func splitJobRetryPhaseAllowed(phase SplitJobPhase) bool { + switch phase { + case SplitJobPhaseBackfill, SplitJobPhaseFence, SplitJobPhaseDeltaCopy, SplitJobPhaseCutover, SplitJobPhaseCleanup: + return true + case SplitJobPhaseNone, SplitJobPhasePlanned, SplitJobPhaseDone, SplitJobPhaseFailed, SplitJobPhaseAbandoning, SplitJobPhaseAbandoned: + return false + } + return false +} + +func splitJobAbandonFromPhase(job SplitJob) (SplitJobPhase, bool) { + switch job.Phase { + case SplitJobPhasePlanned, SplitJobPhaseBackfill, SplitJobPhaseFence, SplitJobPhaseDeltaCopy: + return job.Phase, true + case SplitJobPhaseFailed: + if splitJobAbandonRetryPhaseAllowed(job.RetryPhase) { + return job.RetryPhase, true + } + case SplitJobPhaseAbandoning: + if splitJobAbandonRetryPhaseAllowed(job.AbandonFromPhase) { + return job.AbandonFromPhase, true + } + case SplitJobPhaseNone, SplitJobPhaseCutover, SplitJobPhaseCleanup, SplitJobPhaseDone, SplitJobPhaseAbandoned: + } + return SplitJobPhaseNone, false +} + +func splitJobAbandonRetryPhaseAllowed(phase SplitJobPhase) bool { + switch phase { + case SplitJobPhasePlanned, SplitJobPhaseBackfill, SplitJobPhaseFence, SplitJobPhaseDeltaCopy: + return true + case SplitJobPhaseNone, SplitJobPhaseCutover, SplitJobPhaseCleanup, SplitJobPhaseDone, SplitJobPhaseFailed, SplitJobPhaseAbandoning, SplitJobPhaseAbandoned: + return false + } + return false +} + +// SplitJobsEquivalent reports whether two split jobs encode to the same +// durable catalog value. +func SplitJobsEquivalent(left, right SplitJob) bool { + leftRaw, leftErr := EncodeSplitJob(left) + rightRaw, rightErr := EncodeSplitJob(right) + if leftErr != nil || rightErr != nil { + return false + } + return bytes.Equal(leftRaw, rightRaw) +} diff --git a/main.go b/main.go index ef663e72e..055327215 100644 --- a/main.go +++ b/main.go @@ -53,6 +53,8 @@ const ( etcdMaxSizePerMsg = 1 << 20 etcdMaxInflightMsg = 1024 defaultTSOBatchSize = 256 + + splitMigrationCapabilityProbeTimeout = 2 * time.Second ) func newRaftFactory(engineType raftEngineType, coldStartObs raftengine.ColdStartObserver) (raftengine.Factory, error) { @@ -498,12 +500,20 @@ func run() error { startKeyVizFlusher(runCtx, eg, sampler) startKeyVizLeaderTermPublisher(runCtx, eg, sampler, runtimes) startMemoryWatchdog(runCtx, eg, cancel) + splitMigrationGate := newSplitMigrationCapabilityGate( + splitMigrationCapabilityPeerSourceForRuntimes(runtimes), + splitMigrationCapabilityProbeTimeout, + nil, + ) distServer := adapter.NewDistributionServer( cfg.engine, distCatalog, adapter.WithDistributionCoordinator(coordinate), adapter.WithDistributionActiveTimestampTracker(readTracker), + adapter.WithDistributionKnownRaftGroups(shardGroupIDs(shardGroups)...), + adapter.WithSplitMigrationCapabilityGate(splitMigrationGate), ) + defaultRuntime := findDefaultGroupRuntime(runtimes, cfg.defaultGroup) startMonitoringCollectors(runCtx, metricsRegistry, runtimes, clock) compactor := kv.NewFSMCompactor( fsmCompactionRuntimes(runtimes), @@ -520,7 +530,6 @@ func run() error { // group), in which case raftadmin.Server skips the pre-step. encryptionConfChangeInterceptor := newEncryptionPreRegister( coordinate, shardGroups[cfg.defaultGroup], encWiring.cache, *encryptionSidecarPath, etcdraftengine.DeriveNodeID) - defaultRuntime := findDefaultGroupRuntime(runtimes, cfg.defaultGroup) rotateOnStartupDeregister, waitRotateOnStartup := installEncryptionRotateOnStartup( runCtx, *encryptionRotateOnStartup, @@ -655,6 +664,142 @@ func cloneRaftServers(in []raftengine.Server) []raftengine.Server { return append([]raftengine.Server(nil), in...) } +func shardGroupIDs(groups map[uint64]*kv.ShardGroup) []uint64 { + ids := make([]uint64, 0, len(groups)) + for id := range groups { + ids = append(ids, id) + } + slices.Sort(ids) + return ids +} + +type splitMigrationCapabilityPeer struct { + ID string + Address string +} + +type splitMigrationCapabilityPeerSource func(context.Context) ([]splitMigrationCapabilityPeer, error) + +type splitMigrationCapabilityProbe func(context.Context, string) error + +func splitMigrationCapabilityPeerSourceForRuntimes(runtimes []*raftGroupRuntime) splitMigrationCapabilityPeerSource { + return func(ctx context.Context) ([]splitMigrationCapabilityPeer, error) { + if len(runtimes) == 0 { + return nil, errors.New("raft group runtimes are not configured") + } + peers := make([]splitMigrationCapabilityPeer, 0) + seen := make(map[string]struct{}) + for _, rt := range runtimes { + if rt == nil { + return nil, errors.New("raft group runtime is not configured") + } + engine := rt.snapshotEngine() + if engine == nil { + return nil, errors.Errorf("raft group %d engine is not configured", rt.spec.id) + } + cfg, err := engine.Configuration(ctx) + if err != nil { + return nil, errors.Wrapf(err, "raft group %d configuration", rt.spec.id) + } + groupPeers := splitMigrationCapabilityPeersFromConfiguration(cfg) + if len(groupPeers) == 0 { + return nil, errors.Errorf("raft group %d configuration has no split migration capability peers", rt.spec.id) + } + for _, peer := range groupPeers { + key := peer.ID + "\x00" + peer.Address + if _, ok := seen[key]; ok { + continue + } + seen[key] = struct{}{} + peers = append(peers, peer) + } + } + return peers, nil + } +} + +func splitMigrationCapabilityPeersFromConfiguration(cfg raftengine.Configuration) []splitMigrationCapabilityPeer { + servers := cfg.Servers + if len(servers) == 0 { + return nil + } + peers := make([]splitMigrationCapabilityPeer, 0, len(servers)) + seen := make(map[string]struct{}, len(servers)) + for _, server := range servers { + key := server.ID + "\x00" + server.Address + if _, ok := seen[key]; ok { + continue + } + seen[key] = struct{}{} + id := server.ID + if id == "" { + id = server.Address + } + peers = append(peers, splitMigrationCapabilityPeer{ + ID: id, + Address: server.Address, + }) + } + return peers +} + +func newSplitMigrationCapabilityGate(source splitMigrationCapabilityPeerSource, timeout time.Duration, probe splitMigrationCapabilityProbe) adapter.SplitMigrationCapabilityGate { + if probe == nil { + probe = probeSplitMigrationCapabilityPeer + } + return func(ctx context.Context) error { + if source == nil { + return status.Error(codes.FailedPrecondition, "split migration capability peers are not configured") + } + peers, err := source(ctx) + if err != nil { + return status.Errorf(codes.FailedPrecondition, "split migration capability peers are not available: %v", err) + } + if len(peers) == 0 { + return status.Error(codes.FailedPrecondition, "split migration capability peers are not configured") + } + for _, peer := range peers { + peerCtx := ctx + cancel := func() {} + if timeout > 0 { + var cancelCtx context.CancelFunc + peerCtx, cancelCtx = context.WithTimeout(ctx, timeout) + cancel = cancelCtx + } + err := probe(peerCtx, peer.Address) + cancel() + if err != nil { + return status.Errorf(codes.FailedPrecondition, "split migration capability peer %s is not ready: %v", peer.ID, err) + } + } + return nil + } +} + +func probeSplitMigrationCapabilityPeer(ctx context.Context, address string) error { + if strings.TrimSpace(address) == "" { + return errors.New("empty split migration capability peer address") + } + conn, err := grpc.NewClient(address, internalutil.GRPCDialOptions()...) + if err != nil { + return errors.Wrapf(err, "dial split migration capability peer %s", address) + } + defer func() { + _ = conn.Close() + }() + resp, err := pb.NewDistributionClient(conn).GetSplitMigrationCapability(ctx, &pb.GetSplitMigrationCapabilityRequest{}) + if err != nil { + return errors.Wrapf(err, "probe split migration capability peer %s", address) + } + if resp == nil || !resp.GetMigrationCapable() { + return errors.New("peer does not advertise split migration capability") + } + if !slices.Contains(resp.GetCapabilities(), adapter.SplitMigrationCapabilityV2) { + return errors.Errorf("peer does not advertise %s", adapter.SplitMigrationCapabilityV2) + } + return nil +} + type runtimeConfig struct { groups []groupSpec defaultGroup uint64 @@ -1832,6 +1977,9 @@ func startupRotationGatedMethod(fullMethod string) bool { pb.Internal_ApplyTargetStagedReadiness_FullMethodName, pb.AdminForward_Forward_FullMethodName, pb.Distribution_SplitRange_FullMethodName, + pb.Distribution_StartSplitMigration_FullMethodName, + pb.Distribution_AbandonSplitJob_FullMethodName, + pb.Distribution_RetrySplitJob_FullMethodName, pb.RaftAdmin_AddVoter_FullMethodName, pb.RaftAdmin_AddLearner_FullMethodName, pb.RaftAdmin_PromoteLearner_FullMethodName, diff --git a/main_catalog_test.go b/main_catalog_test.go index 193a98219..b4227a9ee 100644 --- a/main_catalog_test.go +++ b/main_catalog_test.go @@ -2,11 +2,20 @@ package main import ( "context" + "errors" + "net" "testing" + "time" + "github.com/bootjp/elastickv/adapter" "github.com/bootjp/elastickv/distribution" + "github.com/bootjp/elastickv/internal/raftengine" + pb "github.com/bootjp/elastickv/proto" "github.com/bootjp/elastickv/store" "github.com/stretchr/testify/require" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" ) func TestDistributionCatalogGroupID_UsesCatalogKeyRoute(t *testing.T) { @@ -68,3 +77,242 @@ func TestSetupDistributionCatalog_UsesResolvedCatalogGroup(t *testing.T) { require.NoError(t, err) require.NotNil(t, catalog) } + +func TestSplitMigrationCapabilityGateChecksAllPeers(t *testing.T) { + t.Parallel() + + peers := []splitMigrationCapabilityPeer{ + {ID: "n1", Address: "10.0.0.11:50051"}, + {ID: "n2", Address: "10.0.0.12:50051"}, + } + var probed []string + gate := newSplitMigrationCapabilityGate(staticSplitMigrationCapabilityPeerSource(peers), time.Second, func(_ context.Context, address string) error { + probed = append(probed, address) + return nil + }) + + require.NoError(t, gate(context.Background())) + require.Equal(t, []string{"10.0.0.11:50051", "10.0.0.12:50051"}, probed) +} + +func TestSplitMigrationCapabilityGateFailsClosedWithoutPeers(t *testing.T) { + t.Parallel() + + gate := newSplitMigrationCapabilityGate(nil, time.Second, func(context.Context, string) error { + t.Fatal("probe must not run without peers") + return nil + }) + + err := gate(context.Background()) + require.Error(t, err) + require.Equal(t, codes.FailedPrecondition, status.Code(err)) + require.ErrorContains(t, err, "peers are not configured") +} + +func TestSplitMigrationCapabilityGateFailsClosedWhenPeerMissingCapability(t *testing.T) { + t.Parallel() + + peers := []splitMigrationCapabilityPeer{ + {ID: "n1", Address: "10.0.0.11:50051"}, + {ID: "n2", Address: "10.0.0.12:50051"}, + } + gate := newSplitMigrationCapabilityGate(staticSplitMigrationCapabilityPeerSource(peers), time.Second, func(_ context.Context, address string) error { + if address == "10.0.0.12:50051" { + return status.Error(codes.Unimplemented, "method not found") + } + return nil + }) + + err := gate(context.Background()) + require.Error(t, err) + require.Equal(t, codes.FailedPrecondition, status.Code(err)) + require.ErrorContains(t, err, "n2") + require.ErrorContains(t, err, "method not found") +} + +func TestSplitMigrationCapabilityGateFailsClosedWhenPeerSourceErrors(t *testing.T) { + t.Parallel() + + errPeerSourceUnavailable := errors.New("configuration unavailable") + gate := newSplitMigrationCapabilityGate(func(context.Context) ([]splitMigrationCapabilityPeer, error) { + return nil, errPeerSourceUnavailable + }, time.Second, func(context.Context, string) error { + t.Fatal("probe must not run when peer source fails") + return nil + }) + + err := gate(context.Background()) + require.Error(t, err) + require.Equal(t, codes.FailedPrecondition, status.Code(err)) + require.ErrorContains(t, err, "peers are not available") +} + +func TestSplitMigrationCapabilityPeersFromConfigurationUsesCurrentMembers(t *testing.T) { + t.Parallel() + + cfg := raftengine.Configuration{Servers: []raftengine.Server{ + {Suffrage: "voter", ID: "n1", Address: "10.0.0.11:50051"}, + {Suffrage: "learner", ID: "n2", Address: "10.0.0.12:50051"}, + {Suffrage: "", ID: "n3", Address: "10.0.0.13:50051"}, + {Suffrage: "voter", ID: "n1", Address: "10.0.0.11:50051"}, + }} + + require.Equal(t, []splitMigrationCapabilityPeer{ + {ID: "n1", Address: "10.0.0.11:50051"}, + {ID: "n2", Address: "10.0.0.12:50051"}, + {ID: "n3", Address: "10.0.0.13:50051"}, + }, splitMigrationCapabilityPeersFromConfiguration(cfg)) +} + +func TestSplitMigrationCapabilityPeerSourceForRuntimesChecksAllGroups(t *testing.T) { + t.Parallel() + + runtimes := []*raftGroupRuntime{ + { + spec: groupSpec{id: 1}, + engine: capabilityConfigEngine{cfg: raftengine.Configuration{Servers: []raftengine.Server{ + {Suffrage: "voter", ID: "n1", Address: "10.0.0.11:50051"}, + {Suffrage: "learner", ID: "n2", Address: "10.0.0.12:50051"}, + }}}, + }, + { + spec: groupSpec{id: 2}, + engine: capabilityConfigEngine{cfg: raftengine.Configuration{Servers: []raftengine.Server{ + {Suffrage: "voter", ID: "n1", Address: "10.0.0.11:50051"}, + {Suffrage: "voter", ID: "n3", Address: "10.0.0.13:50051"}, + }}}, + }, + } + + peers, err := splitMigrationCapabilityPeerSourceForRuntimes(runtimes)(context.Background()) + require.NoError(t, err) + require.Equal(t, []splitMigrationCapabilityPeer{ + {ID: "n1", Address: "10.0.0.11:50051"}, + {ID: "n2", Address: "10.0.0.12:50051"}, + {ID: "n3", Address: "10.0.0.13:50051"}, + }, peers) +} + +func TestSplitMigrationCapabilityPeerSourceForRuntimesFailsClosedOnEmptyGroup(t *testing.T) { + t.Parallel() + + runtimes := []*raftGroupRuntime{ + { + spec: groupSpec{id: 1}, + engine: capabilityConfigEngine{cfg: raftengine.Configuration{Servers: []raftengine.Server{ + {Suffrage: "voter", ID: "n1", Address: "10.0.0.11:50051"}, + }}}, + }, + { + spec: groupSpec{id: 2}, + engine: capabilityConfigEngine{cfg: raftengine.Configuration{}}, + }, + } + + peers, err := splitMigrationCapabilityPeerSourceForRuntimes(runtimes)(context.Background()) + require.Error(t, err) + require.Nil(t, peers) + require.ErrorContains(t, err, "raft group 2") + require.ErrorContains(t, err, "no split migration capability peers") +} + +func TestProbeSplitMigrationCapabilityPeerRequiresDocumentedToken(t *testing.T) { + t.Parallel() + + addr := startSplitMigrationCapabilityTestServer(t, &pb.GetSplitMigrationCapabilityResponse{ + MigrationCapable: true, + Capabilities: []string{"split_migration_v2"}, + }) + + err := probeSplitMigrationCapabilityPeer(context.Background(), addr) + require.Error(t, err) + require.ErrorContains(t, err, adapter.SplitMigrationCapabilityV2) +} + +func TestProbeSplitMigrationCapabilityPeerAcceptsDocumentedToken(t *testing.T) { + t.Parallel() + + addr := startSplitMigrationCapabilityTestServer(t, &pb.GetSplitMigrationCapabilityResponse{ + MigrationCapable: true, + Capabilities: []string{adapter.SplitMigrationCapabilityV2}, + }) + + require.NoError(t, probeSplitMigrationCapabilityPeer(context.Background(), addr)) +} + +func staticSplitMigrationCapabilityPeerSource(peers []splitMigrationCapabilityPeer) splitMigrationCapabilityPeerSource { + return func(context.Context) ([]splitMigrationCapabilityPeer, error) { + out := make([]splitMigrationCapabilityPeer, len(peers)) + copy(out, peers) + return out, nil + } +} + +type splitMigrationCapabilityTestServer struct { + pb.UnimplementedDistributionServer + resp *pb.GetSplitMigrationCapabilityResponse +} + +func (s *splitMigrationCapabilityTestServer) GetSplitMigrationCapability(context.Context, *pb.GetSplitMigrationCapabilityRequest) (*pb.GetSplitMigrationCapabilityResponse, error) { + return s.resp, nil +} + +func startSplitMigrationCapabilityTestServer(t *testing.T, resp *pb.GetSplitMigrationCapabilityResponse) string { + t.Helper() + + var listenConfig net.ListenConfig + listener, err := listenConfig.Listen(context.Background(), "tcp", "127.0.0.1:0") + require.NoError(t, err) + grpcServer := grpc.NewServer() + pb.RegisterDistributionServer(grpcServer, &splitMigrationCapabilityTestServer{resp: resp}) + done := make(chan struct{}) + go func() { + defer close(done) + _ = grpcServer.Serve(listener) + }() + t.Cleanup(func() { + grpcServer.Stop() + <-done + }) + return listener.Addr().String() +} + +type capabilityConfigEngine struct { + cfg raftengine.Configuration +} + +func (e capabilityConfigEngine) Propose(context.Context, []byte) (*raftengine.ProposalResult, error) { + return nil, nil +} + +func (e capabilityConfigEngine) ProposeAdmin(context.Context, []byte) (*raftengine.ProposalResult, error) { + return nil, nil +} + +func (e capabilityConfigEngine) State() raftengine.State { + return raftengine.StateFollower +} + +func (e capabilityConfigEngine) Leader() raftengine.LeaderInfo { + return raftengine.LeaderInfo{} +} + +func (e capabilityConfigEngine) VerifyLeader(context.Context) error { + return nil +} + +func (e capabilityConfigEngine) LinearizableRead(context.Context) (uint64, error) { + return 0, nil +} + +func (e capabilityConfigEngine) Status() raftengine.Status { + return raftengine.Status{} +} + +func (e capabilityConfigEngine) Configuration(context.Context) (raftengine.Configuration, error) { + return e.cfg, nil +} + +func (e capabilityConfigEngine) Close() error { + return nil +} diff --git a/main_encryption_rotate_on_startup_test.go b/main_encryption_rotate_on_startup_test.go index 750dc673c..fa53d4a68 100644 --- a/main_encryption_rotate_on_startup_test.go +++ b/main_encryption_rotate_on_startup_test.go @@ -422,6 +422,9 @@ func TestStartupPublicKVGate_BlocksMutatorsUntilReady(t *testing.T) { pb.Internal_ApplyTargetStagedReadiness_FullMethodName, pb.AdminForward_Forward_FullMethodName, pb.Distribution_SplitRange_FullMethodName, + pb.Distribution_StartSplitMigration_FullMethodName, + pb.Distribution_AbandonSplitJob_FullMethodName, + pb.Distribution_RetrySplitJob_FullMethodName, pb.RaftAdmin_AddVoter_FullMethodName, pb.RaftAdmin_AddLearner_FullMethodName, pb.RaftAdmin_PromoteLearner_FullMethodName, diff --git a/proto/distribution.pb.go b/proto/distribution.pb.go index 21d15180c..a4a586aa9 100644 --- a/proto/distribution.pb.go +++ b/proto/distribution.pb.go @@ -1280,6 +1280,94 @@ func (x *StartSplitMigrationResponse) GetJobId() uint64 { return 0 } +type GetSplitMigrationCapabilityRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *GetSplitMigrationCapabilityRequest) Reset() { + *x = GetSplitMigrationCapabilityRequest{} + mi := &file_distribution_proto_msgTypes[13] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *GetSplitMigrationCapabilityRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*GetSplitMigrationCapabilityRequest) ProtoMessage() {} + +func (x *GetSplitMigrationCapabilityRequest) ProtoReflect() protoreflect.Message { + mi := &file_distribution_proto_msgTypes[13] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use GetSplitMigrationCapabilityRequest.ProtoReflect.Descriptor instead. +func (*GetSplitMigrationCapabilityRequest) Descriptor() ([]byte, []int) { + return file_distribution_proto_rawDescGZIP(), []int{13} +} + +type GetSplitMigrationCapabilityResponse struct { + state protoimpl.MessageState `protogen:"open.v1"` + MigrationCapable bool `protobuf:"varint,1,opt,name=migration_capable,json=migrationCapable,proto3" json:"migration_capable,omitempty"` + Capabilities []string `protobuf:"bytes,2,rep,name=capabilities,proto3" json:"capabilities,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *GetSplitMigrationCapabilityResponse) Reset() { + *x = GetSplitMigrationCapabilityResponse{} + mi := &file_distribution_proto_msgTypes[14] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *GetSplitMigrationCapabilityResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*GetSplitMigrationCapabilityResponse) ProtoMessage() {} + +func (x *GetSplitMigrationCapabilityResponse) ProtoReflect() protoreflect.Message { + mi := &file_distribution_proto_msgTypes[14] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use GetSplitMigrationCapabilityResponse.ProtoReflect.Descriptor instead. +func (*GetSplitMigrationCapabilityResponse) Descriptor() ([]byte, []int) { + return file_distribution_proto_rawDescGZIP(), []int{14} +} + +func (x *GetSplitMigrationCapabilityResponse) GetMigrationCapable() bool { + if x != nil { + return x.MigrationCapable + } + return false +} + +func (x *GetSplitMigrationCapabilityResponse) GetCapabilities() []string { + if x != nil { + return x.Capabilities + } + return nil +} + type GetRouteOwnershipRequest struct { state protoimpl.MessageState `protogen:"open.v1"` Key []byte `protobuf:"bytes,1,opt,name=key,proto3" json:"key,omitempty"` @@ -1290,7 +1378,7 @@ type GetRouteOwnershipRequest struct { func (x *GetRouteOwnershipRequest) Reset() { *x = GetRouteOwnershipRequest{} - mi := &file_distribution_proto_msgTypes[13] + mi := &file_distribution_proto_msgTypes[15] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1302,7 +1390,7 @@ func (x *GetRouteOwnershipRequest) String() string { func (*GetRouteOwnershipRequest) ProtoMessage() {} func (x *GetRouteOwnershipRequest) ProtoReflect() protoreflect.Message { - mi := &file_distribution_proto_msgTypes[13] + mi := &file_distribution_proto_msgTypes[15] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1315,7 +1403,7 @@ func (x *GetRouteOwnershipRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use GetRouteOwnershipRequest.ProtoReflect.Descriptor instead. func (*GetRouteOwnershipRequest) Descriptor() ([]byte, []int) { - return file_distribution_proto_rawDescGZIP(), []int{13} + return file_distribution_proto_rawDescGZIP(), []int{15} } func (x *GetRouteOwnershipRequest) GetKey() []byte { @@ -1343,7 +1431,7 @@ type GetRouteOwnershipResponse struct { func (x *GetRouteOwnershipResponse) Reset() { *x = GetRouteOwnershipResponse{} - mi := &file_distribution_proto_msgTypes[14] + mi := &file_distribution_proto_msgTypes[16] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1355,7 +1443,7 @@ func (x *GetRouteOwnershipResponse) String() string { func (*GetRouteOwnershipResponse) ProtoMessage() {} func (x *GetRouteOwnershipResponse) ProtoReflect() protoreflect.Message { - mi := &file_distribution_proto_msgTypes[14] + mi := &file_distribution_proto_msgTypes[16] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1368,7 +1456,7 @@ func (x *GetRouteOwnershipResponse) ProtoReflect() protoreflect.Message { // Deprecated: Use GetRouteOwnershipResponse.ProtoReflect.Descriptor instead. func (*GetRouteOwnershipResponse) Descriptor() ([]byte, []int) { - return file_distribution_proto_rawDescGZIP(), []int{14} + return file_distribution_proto_rawDescGZIP(), []int{16} } func (x *GetRouteOwnershipResponse) GetRoute() *RouteDescriptor { @@ -1403,7 +1491,7 @@ type GetIntersectingRoutesRequest struct { func (x *GetIntersectingRoutesRequest) Reset() { *x = GetIntersectingRoutesRequest{} - mi := &file_distribution_proto_msgTypes[15] + mi := &file_distribution_proto_msgTypes[17] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1415,7 +1503,7 @@ func (x *GetIntersectingRoutesRequest) String() string { func (*GetIntersectingRoutesRequest) ProtoMessage() {} func (x *GetIntersectingRoutesRequest) ProtoReflect() protoreflect.Message { - mi := &file_distribution_proto_msgTypes[15] + mi := &file_distribution_proto_msgTypes[17] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1428,7 +1516,7 @@ func (x *GetIntersectingRoutesRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use GetIntersectingRoutesRequest.ProtoReflect.Descriptor instead. func (*GetIntersectingRoutesRequest) Descriptor() ([]byte, []int) { - return file_distribution_proto_rawDescGZIP(), []int{15} + return file_distribution_proto_rawDescGZIP(), []int{17} } func (x *GetIntersectingRoutesRequest) GetStart() []byte { @@ -1462,7 +1550,7 @@ type GetIntersectingRoutesResponse struct { func (x *GetIntersectingRoutesResponse) Reset() { *x = GetIntersectingRoutesResponse{} - mi := &file_distribution_proto_msgTypes[16] + mi := &file_distribution_proto_msgTypes[18] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1474,7 +1562,7 @@ func (x *GetIntersectingRoutesResponse) String() string { func (*GetIntersectingRoutesResponse) ProtoMessage() {} func (x *GetIntersectingRoutesResponse) ProtoReflect() protoreflect.Message { - mi := &file_distribution_proto_msgTypes[16] + mi := &file_distribution_proto_msgTypes[18] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1487,7 +1575,7 @@ func (x *GetIntersectingRoutesResponse) ProtoReflect() protoreflect.Message { // Deprecated: Use GetIntersectingRoutesResponse.ProtoReflect.Descriptor instead. func (*GetIntersectingRoutesResponse) Descriptor() ([]byte, []int) { - return file_distribution_proto_rawDescGZIP(), []int{16} + return file_distribution_proto_rawDescGZIP(), []int{18} } func (x *GetIntersectingRoutesResponse) GetRoutes() []*RouteDescriptor { @@ -1513,7 +1601,7 @@ type GetSplitJobRequest struct { func (x *GetSplitJobRequest) Reset() { *x = GetSplitJobRequest{} - mi := &file_distribution_proto_msgTypes[17] + mi := &file_distribution_proto_msgTypes[19] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1525,7 +1613,7 @@ func (x *GetSplitJobRequest) String() string { func (*GetSplitJobRequest) ProtoMessage() {} func (x *GetSplitJobRequest) ProtoReflect() protoreflect.Message { - mi := &file_distribution_proto_msgTypes[17] + mi := &file_distribution_proto_msgTypes[19] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1538,7 +1626,7 @@ func (x *GetSplitJobRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use GetSplitJobRequest.ProtoReflect.Descriptor instead. func (*GetSplitJobRequest) Descriptor() ([]byte, []int) { - return file_distribution_proto_rawDescGZIP(), []int{17} + return file_distribution_proto_rawDescGZIP(), []int{19} } func (x *GetSplitJobRequest) GetJobId() uint64 { @@ -1557,7 +1645,7 @@ type GetSplitJobResponse struct { func (x *GetSplitJobResponse) Reset() { *x = GetSplitJobResponse{} - mi := &file_distribution_proto_msgTypes[18] + mi := &file_distribution_proto_msgTypes[20] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1569,7 +1657,7 @@ func (x *GetSplitJobResponse) String() string { func (*GetSplitJobResponse) ProtoMessage() {} func (x *GetSplitJobResponse) ProtoReflect() protoreflect.Message { - mi := &file_distribution_proto_msgTypes[18] + mi := &file_distribution_proto_msgTypes[20] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1582,7 +1670,7 @@ func (x *GetSplitJobResponse) ProtoReflect() protoreflect.Message { // Deprecated: Use GetSplitJobResponse.ProtoReflect.Descriptor instead. func (*GetSplitJobResponse) Descriptor() ([]byte, []int) { - return file_distribution_proto_rawDescGZIP(), []int{18} + return file_distribution_proto_rawDescGZIP(), []int{20} } func (x *GetSplitJobResponse) GetJob() *SplitJob { @@ -1603,7 +1691,7 @@ type ListSplitJobsRequest struct { func (x *ListSplitJobsRequest) Reset() { *x = ListSplitJobsRequest{} - mi := &file_distribution_proto_msgTypes[19] + mi := &file_distribution_proto_msgTypes[21] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1615,7 +1703,7 @@ func (x *ListSplitJobsRequest) String() string { func (*ListSplitJobsRequest) ProtoMessage() {} func (x *ListSplitJobsRequest) ProtoReflect() protoreflect.Message { - mi := &file_distribution_proto_msgTypes[19] + mi := &file_distribution_proto_msgTypes[21] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1628,7 +1716,7 @@ func (x *ListSplitJobsRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use ListSplitJobsRequest.ProtoReflect.Descriptor instead. func (*ListSplitJobsRequest) Descriptor() ([]byte, []int) { - return file_distribution_proto_rawDescGZIP(), []int{19} + return file_distribution_proto_rawDescGZIP(), []int{21} } func (x *ListSplitJobsRequest) GetSinceTerminalAtMs() uint64 { @@ -1662,7 +1750,7 @@ type ListSplitJobsResponse struct { func (x *ListSplitJobsResponse) Reset() { *x = ListSplitJobsResponse{} - mi := &file_distribution_proto_msgTypes[20] + mi := &file_distribution_proto_msgTypes[22] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1674,7 +1762,7 @@ func (x *ListSplitJobsResponse) String() string { func (*ListSplitJobsResponse) ProtoMessage() {} func (x *ListSplitJobsResponse) ProtoReflect() protoreflect.Message { - mi := &file_distribution_proto_msgTypes[20] + mi := &file_distribution_proto_msgTypes[22] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1687,7 +1775,7 @@ func (x *ListSplitJobsResponse) ProtoReflect() protoreflect.Message { // Deprecated: Use ListSplitJobsResponse.ProtoReflect.Descriptor instead. func (*ListSplitJobsResponse) Descriptor() ([]byte, []int) { - return file_distribution_proto_rawDescGZIP(), []int{20} + return file_distribution_proto_rawDescGZIP(), []int{22} } func (x *ListSplitJobsResponse) GetJobs() []*SplitJob { @@ -1713,7 +1801,7 @@ type AbandonSplitJobRequest struct { func (x *AbandonSplitJobRequest) Reset() { *x = AbandonSplitJobRequest{} - mi := &file_distribution_proto_msgTypes[21] + mi := &file_distribution_proto_msgTypes[23] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1725,7 +1813,7 @@ func (x *AbandonSplitJobRequest) String() string { func (*AbandonSplitJobRequest) ProtoMessage() {} func (x *AbandonSplitJobRequest) ProtoReflect() protoreflect.Message { - mi := &file_distribution_proto_msgTypes[21] + mi := &file_distribution_proto_msgTypes[23] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1738,7 +1826,7 @@ func (x *AbandonSplitJobRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use AbandonSplitJobRequest.ProtoReflect.Descriptor instead. func (*AbandonSplitJobRequest) Descriptor() ([]byte, []int) { - return file_distribution_proto_rawDescGZIP(), []int{21} + return file_distribution_proto_rawDescGZIP(), []int{23} } func (x *AbandonSplitJobRequest) GetJobId() uint64 { @@ -1756,7 +1844,7 @@ type AbandonSplitJobResponse struct { func (x *AbandonSplitJobResponse) Reset() { *x = AbandonSplitJobResponse{} - mi := &file_distribution_proto_msgTypes[22] + mi := &file_distribution_proto_msgTypes[24] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1768,7 +1856,7 @@ func (x *AbandonSplitJobResponse) String() string { func (*AbandonSplitJobResponse) ProtoMessage() {} func (x *AbandonSplitJobResponse) ProtoReflect() protoreflect.Message { - mi := &file_distribution_proto_msgTypes[22] + mi := &file_distribution_proto_msgTypes[24] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1781,7 +1869,7 @@ func (x *AbandonSplitJobResponse) ProtoReflect() protoreflect.Message { // Deprecated: Use AbandonSplitJobResponse.ProtoReflect.Descriptor instead. func (*AbandonSplitJobResponse) Descriptor() ([]byte, []int) { - return file_distribution_proto_rawDescGZIP(), []int{22} + return file_distribution_proto_rawDescGZIP(), []int{24} } type RetrySplitJobRequest struct { @@ -1793,7 +1881,7 @@ type RetrySplitJobRequest struct { func (x *RetrySplitJobRequest) Reset() { *x = RetrySplitJobRequest{} - mi := &file_distribution_proto_msgTypes[23] + mi := &file_distribution_proto_msgTypes[25] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1805,7 +1893,7 @@ func (x *RetrySplitJobRequest) String() string { func (*RetrySplitJobRequest) ProtoMessage() {} func (x *RetrySplitJobRequest) ProtoReflect() protoreflect.Message { - mi := &file_distribution_proto_msgTypes[23] + mi := &file_distribution_proto_msgTypes[25] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1818,7 +1906,7 @@ func (x *RetrySplitJobRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use RetrySplitJobRequest.ProtoReflect.Descriptor instead. func (*RetrySplitJobRequest) Descriptor() ([]byte, []int) { - return file_distribution_proto_rawDescGZIP(), []int{23} + return file_distribution_proto_rawDescGZIP(), []int{25} } func (x *RetrySplitJobRequest) GetJobId() uint64 { @@ -1836,7 +1924,7 @@ type RetrySplitJobResponse struct { func (x *RetrySplitJobResponse) Reset() { *x = RetrySplitJobResponse{} - mi := &file_distribution_proto_msgTypes[24] + mi := &file_distribution_proto_msgTypes[26] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1848,7 +1936,7 @@ func (x *RetrySplitJobResponse) String() string { func (*RetrySplitJobResponse) ProtoMessage() {} func (x *RetrySplitJobResponse) ProtoReflect() protoreflect.Message { - mi := &file_distribution_proto_msgTypes[24] + mi := &file_distribution_proto_msgTypes[26] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1861,7 +1949,7 @@ func (x *RetrySplitJobResponse) ProtoReflect() protoreflect.Message { // Deprecated: Use RetrySplitJobResponse.ProtoReflect.Descriptor instead. func (*RetrySplitJobResponse) Descriptor() ([]byte, []int) { - return file_distribution_proto_rawDescGZIP(), []int{24} + return file_distribution_proto_rawDescGZIP(), []int{26} } var File_distribution_proto protoreflect.FileDescriptor @@ -1960,7 +2048,11 @@ const file_distribution_proto_rawDesc = "" + "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"]\n" + "\x1bStartSplitMigrationResponse\x12'\n" + "\x0fcatalog_version\x18\x01 \x01(\x04R\x0ecatalogVersion\x12\x15\n" + - "\x06job_id\x18\x02 \x01(\x04R\x05jobId\"U\n" + + "\x06job_id\x18\x02 \x01(\x04R\x05jobId\"$\n" + + "\"GetSplitMigrationCapabilityRequest\"v\n" + + "#GetSplitMigrationCapabilityResponse\x12+\n" + + "\x11migration_capable\x18\x01 \x01(\bR\x10migrationCapable\x12\"\n" + + "\fcapabilities\x18\x02 \x03(\tR\fcapabilities\"U\n" + "\x18GetRouteOwnershipRequest\x12\x10\n" + "\x03key\x18\x01 \x01(\fR\x03key\x12'\n" + "\x0fcatalog_version\x18\x02 \x01(\x04R\x0ecatalogVersion\"\x82\x01\n" + @@ -2021,7 +2113,7 @@ const file_distribution_proto_rawDesc = "" + "\x13SplitJobExportPhase\x12\x1f\n" + "\x1bSPLIT_JOB_EXPORT_PHASE_NONE\x10\x00\x12#\n" + "\x1fSPLIT_JOB_EXPORT_PHASE_BACKFILL\x10\x01\x12%\n" + - "!SPLIT_JOB_EXPORT_PHASE_DELTA_COPY\x10\x022\xf6\x05\n" + + "!SPLIT_JOB_EXPORT_PHASE_DELTA_COPY\x10\x022\xe2\x06\n" + "\fDistribution\x121\n" + "\bGetRoute\x12\x10.GetRouteRequest\x1a\x11.GetRouteResponse\"\x00\x12=\n" + "\fGetTimestamp\x12\x14.GetTimestampRequest\x1a\x15.GetTimestampResponse\"\x00\x127\n" + @@ -2029,7 +2121,8 @@ const file_distribution_proto_rawDesc = "" + "ListRoutes\x12\x12.ListRoutesRequest\x1a\x13.ListRoutesResponse\"\x00\x127\n" + "\n" + "SplitRange\x12\x12.SplitRangeRequest\x1a\x13.SplitRangeResponse\"\x00\x12R\n" + - "\x13StartSplitMigration\x12\x1b.StartSplitMigrationRequest\x1a\x1c.StartSplitMigrationResponse\"\x00\x12L\n" + + "\x13StartSplitMigration\x12\x1b.StartSplitMigrationRequest\x1a\x1c.StartSplitMigrationResponse\"\x00\x12j\n" + + "\x1bGetSplitMigrationCapability\x12#.GetSplitMigrationCapabilityRequest\x1a$.GetSplitMigrationCapabilityResponse\"\x00\x12L\n" + "\x11GetRouteOwnership\x12\x19.GetRouteOwnershipRequest\x1a\x1a.GetRouteOwnershipResponse\"\x00\x12X\n" + "\x15GetIntersectingRoutes\x12\x1d.GetIntersectingRoutesRequest\x1a\x1e.GetIntersectingRoutesResponse\"\x00\x12:\n" + "\vGetSplitJob\x12\x13.GetSplitJobRequest\x1a\x14.GetSplitJobResponse\"\x00\x12@\n" + @@ -2050,38 +2143,40 @@ func file_distribution_proto_rawDescGZIP() []byte { } var file_distribution_proto_enumTypes = make([]protoimpl.EnumInfo, 4) -var file_distribution_proto_msgTypes = make([]protoimpl.MessageInfo, 26) +var file_distribution_proto_msgTypes = make([]protoimpl.MessageInfo, 28) var file_distribution_proto_goTypes = []any{ - (RouteState)(0), // 0: RouteState - (SplitJobPhase)(0), // 1: SplitJobPhase - (SplitJobBarrierState)(0), // 2: SplitJobBarrierState - (SplitJobExportPhase)(0), // 3: SplitJobExportPhase - (*GetRouteRequest)(nil), // 4: GetRouteRequest - (*GetRouteResponse)(nil), // 5: GetRouteResponse - (*GetTimestampRequest)(nil), // 6: GetTimestampRequest - (*GetTimestampResponse)(nil), // 7: GetTimestampResponse - (*RouteDescriptor)(nil), // 8: RouteDescriptor - (*SplitJobBracketProgress)(nil), // 9: SplitJobBracketProgress - (*SplitJob)(nil), // 10: SplitJob - (*ListRoutesRequest)(nil), // 11: ListRoutesRequest - (*ListRoutesResponse)(nil), // 12: ListRoutesResponse - (*SplitRangeRequest)(nil), // 13: SplitRangeRequest - (*SplitRangeResponse)(nil), // 14: SplitRangeResponse - (*StartSplitMigrationRequest)(nil), // 15: StartSplitMigrationRequest - (*StartSplitMigrationResponse)(nil), // 16: StartSplitMigrationResponse - (*GetRouteOwnershipRequest)(nil), // 17: GetRouteOwnershipRequest - (*GetRouteOwnershipResponse)(nil), // 18: GetRouteOwnershipResponse - (*GetIntersectingRoutesRequest)(nil), // 19: GetIntersectingRoutesRequest - (*GetIntersectingRoutesResponse)(nil), // 20: GetIntersectingRoutesResponse - (*GetSplitJobRequest)(nil), // 21: GetSplitJobRequest - (*GetSplitJobResponse)(nil), // 22: GetSplitJobResponse - (*ListSplitJobsRequest)(nil), // 23: ListSplitJobsRequest - (*ListSplitJobsResponse)(nil), // 24: ListSplitJobsResponse - (*AbandonSplitJobRequest)(nil), // 25: AbandonSplitJobRequest - (*AbandonSplitJobResponse)(nil), // 26: AbandonSplitJobResponse - (*RetrySplitJobRequest)(nil), // 27: RetrySplitJobRequest - (*RetrySplitJobResponse)(nil), // 28: RetrySplitJobResponse - nil, // 29: StartSplitMigrationRequest.OptionsEntry + (RouteState)(0), // 0: RouteState + (SplitJobPhase)(0), // 1: SplitJobPhase + (SplitJobBarrierState)(0), // 2: SplitJobBarrierState + (SplitJobExportPhase)(0), // 3: SplitJobExportPhase + (*GetRouteRequest)(nil), // 4: GetRouteRequest + (*GetRouteResponse)(nil), // 5: GetRouteResponse + (*GetTimestampRequest)(nil), // 6: GetTimestampRequest + (*GetTimestampResponse)(nil), // 7: GetTimestampResponse + (*RouteDescriptor)(nil), // 8: RouteDescriptor + (*SplitJobBracketProgress)(nil), // 9: SplitJobBracketProgress + (*SplitJob)(nil), // 10: SplitJob + (*ListRoutesRequest)(nil), // 11: ListRoutesRequest + (*ListRoutesResponse)(nil), // 12: ListRoutesResponse + (*SplitRangeRequest)(nil), // 13: SplitRangeRequest + (*SplitRangeResponse)(nil), // 14: SplitRangeResponse + (*StartSplitMigrationRequest)(nil), // 15: StartSplitMigrationRequest + (*StartSplitMigrationResponse)(nil), // 16: StartSplitMigrationResponse + (*GetSplitMigrationCapabilityRequest)(nil), // 17: GetSplitMigrationCapabilityRequest + (*GetSplitMigrationCapabilityResponse)(nil), // 18: GetSplitMigrationCapabilityResponse + (*GetRouteOwnershipRequest)(nil), // 19: GetRouteOwnershipRequest + (*GetRouteOwnershipResponse)(nil), // 20: GetRouteOwnershipResponse + (*GetIntersectingRoutesRequest)(nil), // 21: GetIntersectingRoutesRequest + (*GetIntersectingRoutesResponse)(nil), // 22: GetIntersectingRoutesResponse + (*GetSplitJobRequest)(nil), // 23: GetSplitJobRequest + (*GetSplitJobResponse)(nil), // 24: GetSplitJobResponse + (*ListSplitJobsRequest)(nil), // 25: ListSplitJobsRequest + (*ListSplitJobsResponse)(nil), // 26: ListSplitJobsResponse + (*AbandonSplitJobRequest)(nil), // 27: AbandonSplitJobRequest + (*AbandonSplitJobResponse)(nil), // 28: AbandonSplitJobResponse + (*RetrySplitJobRequest)(nil), // 29: RetrySplitJobRequest + (*RetrySplitJobResponse)(nil), // 30: RetrySplitJobResponse + nil, // 31: StartSplitMigrationRequest.OptionsEntry } var file_distribution_proto_depIdxs = []int32{ 0, // 0: RouteDescriptor.state:type_name -> RouteState @@ -2095,7 +2190,7 @@ var file_distribution_proto_depIdxs = []int32{ 8, // 8: ListRoutesResponse.routes:type_name -> RouteDescriptor 8, // 9: SplitRangeResponse.left:type_name -> RouteDescriptor 8, // 10: SplitRangeResponse.right:type_name -> RouteDescriptor - 29, // 11: StartSplitMigrationRequest.options:type_name -> StartSplitMigrationRequest.OptionsEntry + 31, // 11: StartSplitMigrationRequest.options:type_name -> StartSplitMigrationRequest.OptionsEntry 8, // 12: GetRouteOwnershipResponse.route:type_name -> RouteDescriptor 8, // 13: GetIntersectingRoutesResponse.routes:type_name -> RouteDescriptor 10, // 14: GetSplitJobResponse.job:type_name -> SplitJob @@ -2105,25 +2200,27 @@ var file_distribution_proto_depIdxs = []int32{ 11, // 18: Distribution.ListRoutes:input_type -> ListRoutesRequest 13, // 19: Distribution.SplitRange:input_type -> SplitRangeRequest 15, // 20: Distribution.StartSplitMigration:input_type -> StartSplitMigrationRequest - 17, // 21: Distribution.GetRouteOwnership:input_type -> GetRouteOwnershipRequest - 19, // 22: Distribution.GetIntersectingRoutes:input_type -> GetIntersectingRoutesRequest - 21, // 23: Distribution.GetSplitJob:input_type -> GetSplitJobRequest - 23, // 24: Distribution.ListSplitJobs:input_type -> ListSplitJobsRequest - 25, // 25: Distribution.AbandonSplitJob:input_type -> AbandonSplitJobRequest - 27, // 26: Distribution.RetrySplitJob:input_type -> RetrySplitJobRequest - 5, // 27: Distribution.GetRoute:output_type -> GetRouteResponse - 7, // 28: Distribution.GetTimestamp:output_type -> GetTimestampResponse - 12, // 29: Distribution.ListRoutes:output_type -> ListRoutesResponse - 14, // 30: Distribution.SplitRange:output_type -> SplitRangeResponse - 16, // 31: Distribution.StartSplitMigration:output_type -> StartSplitMigrationResponse - 18, // 32: Distribution.GetRouteOwnership:output_type -> GetRouteOwnershipResponse - 20, // 33: Distribution.GetIntersectingRoutes:output_type -> GetIntersectingRoutesResponse - 22, // 34: Distribution.GetSplitJob:output_type -> GetSplitJobResponse - 24, // 35: Distribution.ListSplitJobs:output_type -> ListSplitJobsResponse - 26, // 36: Distribution.AbandonSplitJob:output_type -> AbandonSplitJobResponse - 28, // 37: Distribution.RetrySplitJob:output_type -> RetrySplitJobResponse - 27, // [27:38] is the sub-list for method output_type - 16, // [16:27] is the sub-list for method input_type + 17, // 21: Distribution.GetSplitMigrationCapability:input_type -> GetSplitMigrationCapabilityRequest + 19, // 22: Distribution.GetRouteOwnership:input_type -> GetRouteOwnershipRequest + 21, // 23: Distribution.GetIntersectingRoutes:input_type -> GetIntersectingRoutesRequest + 23, // 24: Distribution.GetSplitJob:input_type -> GetSplitJobRequest + 25, // 25: Distribution.ListSplitJobs:input_type -> ListSplitJobsRequest + 27, // 26: Distribution.AbandonSplitJob:input_type -> AbandonSplitJobRequest + 29, // 27: Distribution.RetrySplitJob:input_type -> RetrySplitJobRequest + 5, // 28: Distribution.GetRoute:output_type -> GetRouteResponse + 7, // 29: Distribution.GetTimestamp:output_type -> GetTimestampResponse + 12, // 30: Distribution.ListRoutes:output_type -> ListRoutesResponse + 14, // 31: Distribution.SplitRange:output_type -> SplitRangeResponse + 16, // 32: Distribution.StartSplitMigration:output_type -> StartSplitMigrationResponse + 18, // 33: Distribution.GetSplitMigrationCapability:output_type -> GetSplitMigrationCapabilityResponse + 20, // 34: Distribution.GetRouteOwnership:output_type -> GetRouteOwnershipResponse + 22, // 35: Distribution.GetIntersectingRoutes:output_type -> GetIntersectingRoutesResponse + 24, // 36: Distribution.GetSplitJob:output_type -> GetSplitJobResponse + 26, // 37: Distribution.ListSplitJobs:output_type -> ListSplitJobsResponse + 28, // 38: Distribution.AbandonSplitJob:output_type -> AbandonSplitJobResponse + 30, // 39: Distribution.RetrySplitJob:output_type -> RetrySplitJobResponse + 28, // [28:40] is the sub-list for method output_type + 16, // [16:28] is the sub-list for method input_type 16, // [16:16] is the sub-list for extension type_name 16, // [16:16] is the sub-list for extension extendee 0, // [0:16] is the sub-list for field type_name @@ -2140,7 +2237,7 @@ func file_distribution_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_distribution_proto_rawDesc), len(file_distribution_proto_rawDesc)), NumEnums: 4, - NumMessages: 26, + NumMessages: 28, NumExtensions: 0, NumServices: 1, }, diff --git a/proto/distribution.proto b/proto/distribution.proto index eb9746f18..4f65163b0 100644 --- a/proto/distribution.proto +++ b/proto/distribution.proto @@ -8,6 +8,7 @@ service Distribution { rpc ListRoutes (ListRoutesRequest) returns (ListRoutesResponse) {} rpc SplitRange (SplitRangeRequest) returns (SplitRangeResponse) {} rpc StartSplitMigration (StartSplitMigrationRequest) returns (StartSplitMigrationResponse) {} + rpc GetSplitMigrationCapability (GetSplitMigrationCapabilityRequest) returns (GetSplitMigrationCapabilityResponse) {} rpc GetRouteOwnership (GetRouteOwnershipRequest) returns (GetRouteOwnershipResponse) {} rpc GetIntersectingRoutes (GetIntersectingRoutesRequest) returns (GetIntersectingRoutesResponse) {} rpc GetSplitJob (GetSplitJobRequest) returns (GetSplitJobResponse) {} @@ -160,6 +161,13 @@ message StartSplitMigrationResponse { uint64 job_id = 2; } +message GetSplitMigrationCapabilityRequest {} + +message GetSplitMigrationCapabilityResponse { + bool migration_capable = 1; + repeated string capabilities = 2; +} + message GetRouteOwnershipRequest { bytes key = 1; uint64 catalog_version = 2; diff --git a/proto/distribution_grpc.pb.go b/proto/distribution_grpc.pb.go index 66ebf8511..ec02eabee 100644 --- a/proto/distribution_grpc.pb.go +++ b/proto/distribution_grpc.pb.go @@ -19,17 +19,18 @@ import ( const _ = grpc.SupportPackageIsVersion9 const ( - Distribution_GetRoute_FullMethodName = "/Distribution/GetRoute" - Distribution_GetTimestamp_FullMethodName = "/Distribution/GetTimestamp" - Distribution_ListRoutes_FullMethodName = "/Distribution/ListRoutes" - Distribution_SplitRange_FullMethodName = "/Distribution/SplitRange" - Distribution_StartSplitMigration_FullMethodName = "/Distribution/StartSplitMigration" - Distribution_GetRouteOwnership_FullMethodName = "/Distribution/GetRouteOwnership" - Distribution_GetIntersectingRoutes_FullMethodName = "/Distribution/GetIntersectingRoutes" - Distribution_GetSplitJob_FullMethodName = "/Distribution/GetSplitJob" - Distribution_ListSplitJobs_FullMethodName = "/Distribution/ListSplitJobs" - Distribution_AbandonSplitJob_FullMethodName = "/Distribution/AbandonSplitJob" - Distribution_RetrySplitJob_FullMethodName = "/Distribution/RetrySplitJob" + Distribution_GetRoute_FullMethodName = "/Distribution/GetRoute" + Distribution_GetTimestamp_FullMethodName = "/Distribution/GetTimestamp" + Distribution_ListRoutes_FullMethodName = "/Distribution/ListRoutes" + Distribution_SplitRange_FullMethodName = "/Distribution/SplitRange" + Distribution_StartSplitMigration_FullMethodName = "/Distribution/StartSplitMigration" + Distribution_GetSplitMigrationCapability_FullMethodName = "/Distribution/GetSplitMigrationCapability" + Distribution_GetRouteOwnership_FullMethodName = "/Distribution/GetRouteOwnership" + Distribution_GetIntersectingRoutes_FullMethodName = "/Distribution/GetIntersectingRoutes" + Distribution_GetSplitJob_FullMethodName = "/Distribution/GetSplitJob" + Distribution_ListSplitJobs_FullMethodName = "/Distribution/ListSplitJobs" + Distribution_AbandonSplitJob_FullMethodName = "/Distribution/AbandonSplitJob" + Distribution_RetrySplitJob_FullMethodName = "/Distribution/RetrySplitJob" ) // DistributionClient is the client API for Distribution service. @@ -41,6 +42,7 @@ type DistributionClient interface { ListRoutes(ctx context.Context, in *ListRoutesRequest, opts ...grpc.CallOption) (*ListRoutesResponse, error) SplitRange(ctx context.Context, in *SplitRangeRequest, opts ...grpc.CallOption) (*SplitRangeResponse, error) StartSplitMigration(ctx context.Context, in *StartSplitMigrationRequest, opts ...grpc.CallOption) (*StartSplitMigrationResponse, error) + GetSplitMigrationCapability(ctx context.Context, in *GetSplitMigrationCapabilityRequest, opts ...grpc.CallOption) (*GetSplitMigrationCapabilityResponse, error) GetRouteOwnership(ctx context.Context, in *GetRouteOwnershipRequest, opts ...grpc.CallOption) (*GetRouteOwnershipResponse, error) GetIntersectingRoutes(ctx context.Context, in *GetIntersectingRoutesRequest, opts ...grpc.CallOption) (*GetIntersectingRoutesResponse, error) GetSplitJob(ctx context.Context, in *GetSplitJobRequest, opts ...grpc.CallOption) (*GetSplitJobResponse, error) @@ -107,6 +109,16 @@ func (c *distributionClient) StartSplitMigration(ctx context.Context, in *StartS return out, nil } +func (c *distributionClient) GetSplitMigrationCapability(ctx context.Context, in *GetSplitMigrationCapabilityRequest, opts ...grpc.CallOption) (*GetSplitMigrationCapabilityResponse, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(GetSplitMigrationCapabilityResponse) + err := c.cc.Invoke(ctx, Distribution_GetSplitMigrationCapability_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + func (c *distributionClient) GetRouteOwnership(ctx context.Context, in *GetRouteOwnershipRequest, opts ...grpc.CallOption) (*GetRouteOwnershipResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(GetRouteOwnershipResponse) @@ -176,6 +188,7 @@ type DistributionServer interface { ListRoutes(context.Context, *ListRoutesRequest) (*ListRoutesResponse, error) SplitRange(context.Context, *SplitRangeRequest) (*SplitRangeResponse, error) StartSplitMigration(context.Context, *StartSplitMigrationRequest) (*StartSplitMigrationResponse, error) + GetSplitMigrationCapability(context.Context, *GetSplitMigrationCapabilityRequest) (*GetSplitMigrationCapabilityResponse, error) GetRouteOwnership(context.Context, *GetRouteOwnershipRequest) (*GetRouteOwnershipResponse, error) GetIntersectingRoutes(context.Context, *GetIntersectingRoutesRequest) (*GetIntersectingRoutesResponse, error) GetSplitJob(context.Context, *GetSplitJobRequest) (*GetSplitJobResponse, error) @@ -207,6 +220,9 @@ func (UnimplementedDistributionServer) SplitRange(context.Context, *SplitRangeRe func (UnimplementedDistributionServer) StartSplitMigration(context.Context, *StartSplitMigrationRequest) (*StartSplitMigrationResponse, error) { return nil, status.Error(codes.Unimplemented, "method StartSplitMigration not implemented") } +func (UnimplementedDistributionServer) GetSplitMigrationCapability(context.Context, *GetSplitMigrationCapabilityRequest) (*GetSplitMigrationCapabilityResponse, error) { + return nil, status.Error(codes.Unimplemented, "method GetSplitMigrationCapability not implemented") +} func (UnimplementedDistributionServer) GetRouteOwnership(context.Context, *GetRouteOwnershipRequest) (*GetRouteOwnershipResponse, error) { return nil, status.Error(codes.Unimplemented, "method GetRouteOwnership not implemented") } @@ -336,6 +352,24 @@ func _Distribution_StartSplitMigration_Handler(srv interface{}, ctx context.Cont return interceptor(ctx, in, info, handler) } +func _Distribution_GetSplitMigrationCapability_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(GetSplitMigrationCapabilityRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(DistributionServer).GetSplitMigrationCapability(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: Distribution_GetSplitMigrationCapability_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(DistributionServer).GetSplitMigrationCapability(ctx, req.(*GetSplitMigrationCapabilityRequest)) + } + return interceptor(ctx, in, info, handler) +} + func _Distribution_GetRouteOwnership_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(GetRouteOwnershipRequest) if err := dec(in); err != nil { @@ -471,6 +505,10 @@ var Distribution_ServiceDesc = grpc.ServiceDesc{ MethodName: "StartSplitMigration", Handler: _Distribution_StartSplitMigration_Handler, }, + { + MethodName: "GetSplitMigrationCapability", + Handler: _Distribution_GetSplitMigrationCapability_Handler, + }, { MethodName: "GetRouteOwnership", Handler: _Distribution_GetRouteOwnership_Handler,