diff --git a/adapter/distribution_milestone1_e2e_test.go b/adapter/distribution_milestone1_e2e_test.go index d85549402..c621db382 100644 --- a/adapter/distribution_milestone1_e2e_test.go +++ b/adapter/distribution_milestone1_e2e_test.go @@ -89,6 +89,12 @@ func TestMilestone1SplitRange_EndToEndRefreshAndDataPath(t *testing.T) { }) require.NoError(t, err) require.Equal(t, uint64(2), splitResp.CatalogVersion) + changes, err := catalog.ChangesSince(ctx, initial.Version, 1) + require.NoError(t, err) + require.Nil(t, changes.Reset) + require.Len(t, changes.Deltas, 1) + require.Equal(t, splitResp.CatalogVersion, changes.Deltas[0].Version) + require.Len(t, changes.Deltas[0].Mutations, 3) require.Eventually(t, func() bool { if engine.Version() != splitResp.CatalogVersion { diff --git a/adapter/distribution_server.go b/adapter/distribution_server.go index 0e92f78de..072294d78 100644 --- a/adapter/distribution_server.go +++ b/adapter/distribution_server.go @@ -13,6 +13,7 @@ import ( "github.com/bootjp/elastickv/internal/fskeys" "github.com/bootjp/elastickv/kv" pb "github.com/bootjp/elastickv/proto" + "github.com/bootjp/elastickv/store" "github.com/cockroachdb/errors" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" @@ -20,13 +21,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 - fsObserver DistributionFilesystemObserver - reloadRetry struct { + mu sync.Mutex + engine *distribution.Engine + catalog *distribution.CatalogStore + coordinator kv.Coordinator + readTracker *kv.ActiveTimestampTracker + watchInterval time.Duration + watchLeader func() bool + fsObserver DistributionFilesystemObserver + reloadRetry struct { attempts int interval time.Duration } @@ -75,6 +78,25 @@ func WithCatalogReloadRetryPolicy(attempts int, interval time.Duration) Distribu } } +// WithCatalogWatchInterval sets how often an idle catalog stream checks for a +// newly committed version. +func WithCatalogWatchInterval(interval time.Duration) DistributionServerOption { + return func(s *DistributionServer) { + if interval > 0 { + s.watchInterval = interval + } + } +} + +// WithCatalogWatchLeaderCheck restricts long-lived catalog streams to the +// current default-group leader. Existing callers without a check retain the +// package-level server behavior used by tests and embedded deployments. +func WithCatalogWatchLeaderCheck(check func() bool) DistributionServerOption { + return func(s *DistributionServer) { + s.watchLeader = check + } +} + const ( childRouteCount = 2 splitMutationOpCount = childRouteCount + 3 @@ -83,23 +105,26 @@ const ( var ( defaultCatalogReloadRetryAttempts = 20 defaultCatalogReloadRetryInterval = 10 * time.Millisecond - - errDistributionCatalogNotConfigured = errors.New("route catalog is not configured") - errDistributionUnknownRoute = errors.New("unknown route") - errDistributionInvalidSplitKey = errors.New("invalid split key") - errDistributionSplitKeyAtBoundary = errors.New("split key at route boundary") - errDistributionCatalogConflict = errors.New("catalog version conflict") - errDistributionRouteIDOverflow = errors.New("route id overflow") - errDistributionNotLeader = errors.New("not leader for distribution catalog") - errDistributionCoordinatorRequired = errors.New("distribution coordinator is not configured") - errDistributionEngineNotConfigured = errors.New("distribution engine is not configured") + defaultCatalogWatchInterval = 100 * time.Millisecond + + errDistributionCatalogNotConfigured = errors.New("route catalog is not configured") + errDistributionUnknownRoute = errors.New("unknown route") + errDistributionInvalidSplitKey = errors.New("invalid split key") + errDistributionSplitKeyAtBoundary = errors.New("split key at route boundary") + errDistributionCatalogConflict = errors.New("catalog version conflict") + errDistributionRouteIDOverflow = errors.New("route id overflow") + errDistributionNotLeader = errors.New("not leader for distribution catalog") + errDistributionCoordinatorRequired = errors.New("distribution coordinator is not configured") + errDistributionEngineNotConfigured = errors.New("distribution engine is not configured") + errDistributionCatalogMutationInvalid = errors.New("catalog store mutation is invalid") ) // NewDistributionServer creates a new server. func NewDistributionServer(e *distribution.Engine, catalog *distribution.CatalogStore, opts ...DistributionServerOption) *DistributionServer { s := &DistributionServer{ - engine: e, - catalog: catalog, + engine: e, + catalog: catalog, + watchInterval: defaultCatalogWatchInterval, } s.reloadRetry.attempts = defaultCatalogReloadRetryAttempts s.reloadRetry.interval = defaultCatalogReloadRetryInterval @@ -148,6 +173,164 @@ func (s *DistributionServer) ListRoutes(ctx context.Context, req *pb.ListRoutesR }, nil } +// GetCatalogCapabilities negotiates the durable delta-watch protocol. +func (s *DistributionServer) GetCatalogCapabilities(ctx context.Context, _ *pb.CatalogCapabilitiesRequest) (*pb.CatalogCapabilitiesResponse, error) { + if s.catalog == nil { + return nil, grpcStatusError(codes.FailedPrecondition, errDistributionCatalogNotConfigured.Error()) + } + version, err := s.catalog.Version(ctx) + if err != nil { + return nil, grpcStatusErrorf(codes.Internal, "load catalog version: %v", err) + } + floor, err := s.catalog.DeltaFloor(ctx) + if err != nil { + return nil, grpcStatusErrorf(codes.Internal, "load catalog delta floor: %v", err) + } + return &pb.CatalogCapabilitiesResponse{ + SupportedProtocolVersions: []uint32{distribution.CatalogWatchProtocolVersion}, + CurrentVersion: version, + OldestDeltaVersion: floor, + MaxBatchSize: distribution.MaxCatalogDeltaBatchSize, + }, nil +} + +// WatchCatalog streams contiguous deltas and emits a snapshot reset when the +// requested reconnect cursor predates retained history. +func (s *DistributionServer) WatchCatalog(req *pb.CatalogWatchRequest, stream pb.Distribution_WatchCatalogServer) error { + config, err := s.catalogWatchConfig(req) + if err != nil { + return err + } + for { + if s.watchLeader != nil && !s.watchLeader() { + return grpcStatusError(codes.Unavailable, "catalog watch requires the catalog-group leader") + } + nextCursor, sent, err := s.sendCatalogChanges(stream, config.cursor, config.batchSize) + if err != nil { + return err + } + config.cursor = nextCursor + if sent { + continue + } + stop, err := waitCatalogStream(stream.Context(), config.interval) + if err != nil { + return err + } + if stop { + return nil + } + } +} + +type catalogWatchConfig struct { + cursor uint64 + batchSize int + interval time.Duration +} + +func (s *DistributionServer) catalogWatchConfig(req *pb.CatalogWatchRequest) (catalogWatchConfig, error) { + if s.catalog == nil { + return catalogWatchConfig{}, grpcStatusError(codes.FailedPrecondition, errDistributionCatalogNotConfigured.Error()) + } + if req.GetProtocolVersion() != distribution.CatalogWatchProtocolVersion { + return catalogWatchConfig{}, grpcStatusErrorf(codes.FailedPrecondition, "unsupported catalog watch protocol %d", req.GetProtocolVersion()) + } + batchSize := int(req.GetMaxBatchSize()) + if batchSize <= 0 { + batchSize = distribution.DefaultCatalogDeltaBatchSize + } + if batchSize > distribution.MaxCatalogDeltaBatchSize { + batchSize = distribution.MaxCatalogDeltaBatchSize + } + interval := s.watchInterval + if interval <= 0 { + interval = defaultCatalogWatchInterval + } + return catalogWatchConfig{cursor: req.GetAfterVersion(), batchSize: batchSize, interval: interval}, nil +} + +func (s *DistributionServer) sendCatalogChanges( + stream pb.Distribution_WatchCatalogServer, + cursor uint64, + batchSize int, +) (uint64, bool, error) { + changes, err := s.catalog.ChangesSince(stream.Context(), cursor, batchSize) + if err != nil { + if errors.Is(err, distribution.ErrCatalogDeltaVersionFuture) { + return cursor, false, grpcStatusError(codes.InvalidArgument, err.Error()) + } + return cursor, false, grpcStatusErrorf(codes.Internal, "read catalog changes: %v", err) + } + if changes.Reset != nil { + return s.sendCatalogReset(stream, changes.Reset) + } + return sendCatalogDeltas(stream, cursor, changes.Deltas) +} + +func (s *DistributionServer) sendCatalogReset( + stream pb.Distribution_WatchCatalogServer, + snapshot *distribution.CatalogSnapshot, +) (uint64, bool, error) { + event := &pb.CatalogWatchEvent{Payload: &pb.CatalogWatchEvent_Snapshot{ + Snapshot: &pb.CatalogSnapshotReset{ + Version: snapshot.Version, + Routes: toProtoRouteDescriptors(snapshot.Routes), + }, + }} + if err := stream.Send(event); err != nil { + return 0, false, errors.WithStack(err) + } + return snapshot.Version, true, nil +} + +func sendCatalogDeltas( + stream pb.Distribution_WatchCatalogServer, + cursor uint64, + deltas []distribution.CatalogDelta, +) (uint64, bool, error) { + for _, delta := range deltas { + event := &pb.CatalogWatchEvent{Payload: &pb.CatalogWatchEvent_Delta{ + Delta: toProtoCatalogDelta(delta), + }} + if err := stream.Send(event); err != nil { + return cursor, false, errors.WithStack(err) + } + cursor = delta.Version + } + return cursor, len(deltas) > 0, nil +} + +func waitCatalogStream(ctx context.Context, interval time.Duration) (bool, error) { + err := waitWithContext(ctx, interval) + if errors.Is(err, context.Canceled) || status.Code(errors.Cause(err)) == codes.Canceled { + return true, nil + } + return false, err +} + +func toProtoCatalogDelta(delta distribution.CatalogDelta) *pb.CatalogDeltaRecord { + mutations := make([]*pb.CatalogDeltaMutation, 0, len(delta.Mutations)) + for _, mutation := range delta.Mutations { + op := pb.CatalogDeltaMutationOp_CATALOG_DELTA_MUTATION_OP_DELETE + var route *pb.RouteDescriptor + if mutation.Op == distribution.CatalogMutationUpsert { + op = pb.CatalogDeltaMutationOp_CATALOG_DELTA_MUTATION_OP_UPSERT + route = toProtoRouteDescriptor(mutation.Route) + } + mutations = append(mutations, &pb.CatalogDeltaMutation{ + Op: op, + RouteId: mutation.RouteID, + Route: route, + }) + } + return &pb.CatalogDeltaRecord{ + PreviousVersion: delta.PreviousVersion, + Version: delta.Version, + Mutations: mutations, + } +} + // SplitRange splits a route into two child routes in the same raft group. func (s *DistributionServer) SplitRange(ctx context.Context, req *pb.SplitRangeRequest) (*pb.SplitRangeResponse, error) { // SplitRange performs a read-modify-write cycle across catalog and engine. @@ -265,15 +448,40 @@ func (s *DistributionServer) saveSplitResultViaCoordinator( return distribution.CatalogSnapshot{}, grpcStatusError(codes.Internal, errDistributionRouteIDOverflow.Error()) } nextRouteID := right.RouteID + 1 + commitTS, err := kv.NextTimestampAfterThrough(ctx, s.coordinator, readTS, "split range: allocate commitTS") + if err != nil { + return distribution.CatalogSnapshot{}, grpcStatusErrorf(codes.Internal, "allocate split commit timestamp: %v", err) + } + left.SplitAtHLC = commitTS + right.SplitAtHLC = commitTS ops, err := buildCatalogSplitOps(parentID, left, right, nextVersion, nextRouteID) if err != nil { return distribution.CatalogSnapshot{}, grpcStatusErrorf(codes.Internal, "build split mutations: %v", err) } + delta := distribution.CatalogDelta{ + PreviousVersion: expectedVersion, + Version: nextVersion, + Mutations: []distribution.CatalogRouteMutation{ + {Op: distribution.CatalogMutationDelete, RouteID: parentID}, + {Op: distribution.CatalogMutationUpsert, RouteID: left.RouteID, Route: left}, + {Op: distribution.CatalogMutationUpsert, RouteID: right.RouteID, Route: right}, + }, + } + deltaMutations, err := s.catalog.BuildDeltaMutationsAt(ctx, readTS, delta) + if err != nil { + return distribution.CatalogSnapshot{}, grpcStatusErrorf(codes.Internal, "build catalog delta mutations: %v", err) + } + deltaOps, err := catalogStoreMutationsToOps(deltaMutations) + if err != nil { + return distribution.CatalogSnapshot{}, grpcStatusErrorf(codes.Internal, "convert catalog delta mutations: %v", err) + } + ops = append(ops, deltaOps...) resp, err := s.coordinator.Dispatch(ctx, &kv.OperationGroup[kv.OP]{ - Elems: ops, - IsTxn: true, - StartTS: readTS, + Elems: ops, + IsTxn: true, + StartTS: readTS, + CommitTS: commitTS, }) if err != nil { return distribution.CatalogSnapshot{}, grpcStatusErrorf(codes.Internal, "commit split mutations: %v", err) @@ -284,6 +492,30 @@ func (s *DistributionServer) saveSplitResultViaCoordinator( return s.loadCatalogSnapshotAtVersion(ctx, resp.CommitTS, nextVersion) } +func catalogStoreMutationsToOps(mutations []*store.KVPairMutation) ([]*kv.Elem[kv.OP], error) { + ops := make([]*kv.Elem[kv.OP], 0, len(mutations)) + for _, mutation := range mutations { + if mutation == nil { + return nil, errors.WithStack(errDistributionCatalogMutationInvalid) + } + var op kv.OP + switch mutation.Op { + case store.OpTypePut: + op = kv.Put + case store.OpTypeDelete: + op = kv.Del + default: + return nil, errors.Wrapf(errDistributionCatalogMutationInvalid, "unknown operation %d", mutation.Op) + } + ops = append(ops, &kv.Elem[kv.OP]{ + Op: op, + Key: distribution.CloneBytes(mutation.Key), + Value: distribution.CloneBytes(mutation.Value), + }) + } + return ops, nil +} + func buildCatalogSplitOps( parentID uint64, left distribution.RouteDescriptor, diff --git a/adapter/distribution_server_delta_test.go b/adapter/distribution_server_delta_test.go new file mode 100644 index 000000000..b9057792c --- /dev/null +++ b/adapter/distribution_server_delta_test.go @@ -0,0 +1,481 @@ +package adapter + +import ( + "bytes" + "context" + "io" + "log/slog" + "net" + "sync/atomic" + "testing" + "time" + + "github.com/bootjp/elastickv/distribution" + pb "github.com/bootjp/elastickv/proto" + "github.com/bootjp/elastickv/store" + "github.com/stretchr/testify/require" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/metadata" + "google.golang.org/grpc/status" + "google.golang.org/grpc/test/bufconn" +) + +func TestDistributionServerCatalogWatchCapabilitiesAndReconnect(t *testing.T) { + t.Parallel() + + ctx := context.Background() + catalog := distribution.NewCatalogStore(store.NewMVCCStore()) + routes := []distribution.RouteDescriptor{ + {RouteID: 1, Start: []byte(""), End: nil, GroupID: 1, State: distribution.RouteStateActive}, + } + first, err := catalog.Save(ctx, 0, routes) + require.NoError(t, err) + server := NewDistributionServer(distribution.NewEngine(), catalog) + + capabilities, err := server.GetCatalogCapabilities(ctx, &pb.CatalogCapabilitiesRequest{}) + require.NoError(t, err) + require.Equal(t, []uint32{distribution.CatalogWatchProtocolVersion}, capabilities.GetSupportedProtocolVersions()) + require.Equal(t, first.Version, capabilities.GetCurrentVersion()) + require.EqualValues(t, 1, capabilities.GetOldestDeltaVersion()) + require.EqualValues(t, distribution.MaxCatalogDeltaBatchSize, capabilities.GetMaxBatchSize()) + + firstStream := newCaptureCatalogWatchStream() + require.NoError(t, server.WatchCatalog(&pb.CatalogWatchRequest{ + ProtocolVersion: distribution.CatalogWatchProtocolVersion, + AfterVersion: 0, + }, firstStream)) + firstEvent := <-firstStream.events + require.Equal(t, first.Version, firstEvent.GetDelta().GetVersion()) + require.EqualValues(t, 0, firstEvent.GetDelta().GetPreviousVersion()) + + second, err := catalog.Save(ctx, first.Version, routes) + require.NoError(t, err) + secondStream := newCaptureCatalogWatchStream() + require.NoError(t, server.WatchCatalog(&pb.CatalogWatchRequest{ + ProtocolVersion: distribution.CatalogWatchProtocolVersion, + AfterVersion: first.Version, + }, secondStream)) + secondEvent := <-secondStream.events + require.Equal(t, second.Version, secondEvent.GetDelta().GetVersion()) + require.Equal(t, first.Version, secondEvent.GetDelta().GetPreviousVersion()) +} + +func TestDistributionServerCatalogWatchRejectsProtocolAndFutureCursor(t *testing.T) { + t.Parallel() + + catalog := distribution.NewCatalogStore(store.NewMVCCStore()) + server := NewDistributionServer(distribution.NewEngine(), catalog) + + err := server.WatchCatalog(&pb.CatalogWatchRequest{ProtocolVersion: 99}, newCaptureCatalogWatchStream()) + require.Equal(t, codes.FailedPrecondition, status.Code(err)) + + err = server.WatchCatalog(&pb.CatalogWatchRequest{ + ProtocolVersion: distribution.CatalogWatchProtocolVersion, + AfterVersion: 1, + }, newCaptureCatalogWatchStream()) + require.Equal(t, codes.InvalidArgument, status.Code(err)) +} + +func TestDistributionServerCatalogWatchRequiresConfiguredLeader(t *testing.T) { + t.Parallel() + + ctx := context.Background() + catalog := distribution.NewCatalogStore(store.NewMVCCStore()) + _, err := catalog.Save(ctx, 0, []distribution.RouteDescriptor{ + {RouteID: 1, Start: []byte(""), End: nil, GroupID: 1, State: distribution.RouteStateActive}, + }) + require.NoError(t, err) + + var leader atomic.Bool + server := NewDistributionServer( + distribution.NewEngine(), + catalog, + WithCatalogWatchLeaderCheck(leader.Load), + ) + req := &pb.CatalogWatchRequest{ProtocolVersion: distribution.CatalogWatchProtocolVersion} + err = server.WatchCatalog(req, newCaptureCatalogWatchStream()) + require.Equal(t, codes.Unavailable, status.Code(err)) + + leader.Store(true) + require.NoError(t, server.WatchCatalog(req, newCaptureCatalogWatchStream())) +} + +func TestDistributionServerCatalogWatchRejectsFollower(t *testing.T) { + t.Parallel() + + server := NewDistributionServer( + distribution.NewEngine(), + distribution.NewCatalogStore(store.NewMVCCStore()), + WithCatalogWatchLeaderCheck(func() bool { return false }), + ) + err := server.WatchCatalog(&pb.CatalogWatchRequest{ + ProtocolVersion: distribution.CatalogWatchProtocolVersion, + }, newCaptureCatalogWatchStream()) + require.Equal(t, codes.Unavailable, status.Code(err)) +} + +func TestDistributionServerCatalogWatchSendsSnapshotReset(t *testing.T) { + t.Parallel() + + ctx := context.Background() + st := store.NewMVCCStore() + catalog := distribution.NewCatalogStore(st) + routes := []distribution.RouteDescriptor{ + {RouteID: 1, Start: []byte(""), End: nil, GroupID: 1, State: distribution.RouteStateActive}, + } + first, err := catalog.Save(ctx, 0, routes) + require.NoError(t, err) + second, err := catalog.Save(ctx, first.Version, routes) + require.NoError(t, err) + + startTS := st.LastCommitTS() + require.NoError(t, st.ApplyMutations(ctx, []*store.KVPairMutation{ + {Op: store.OpTypeDelete, Key: distribution.CatalogDeltaKey(first.Version)}, + {Op: store.OpTypePut, Key: distribution.CatalogDeltaFloorKey(), Value: distribution.EncodeCatalogVersion(second.Version)}, + }, nil, startTS, startTS+1)) + + server := NewDistributionServer(distribution.NewEngine(), catalog) + stream := newCaptureCatalogWatchStream() + require.NoError(t, server.WatchCatalog(&pb.CatalogWatchRequest{ + ProtocolVersion: distribution.CatalogWatchProtocolVersion, + AfterVersion: 0, + }, stream)) + event := <-stream.events + require.Nil(t, event.GetDelta()) + require.Equal(t, second.Version, event.GetSnapshot().GetVersion()) + require.Len(t, event.GetSnapshot().GetRoutes(), 1) +} + +func TestGRPCCatalogWatcherReconnectsFromPublishedVersion(t *testing.T) { + t.Parallel() + + ctx := context.Background() + catalog := distribution.NewCatalogStore(store.NewMVCCStore()) + routes := []distribution.RouteDescriptor{ + {RouteID: 1, Start: []byte(""), End: nil, GroupID: 1, State: distribution.RouteStateActive}, + } + first, err := catalog.Save(ctx, 0, routes) + require.NoError(t, err) + second, err := catalog.Save(ctx, first.Version, routes) + require.NoError(t, err) + + base := NewDistributionServer( + distribution.NewEngine(), + catalog, + WithCatalogWatchInterval(time.Millisecond), + ) + flaky := &flakyCatalogWatchServer{DistributionServer: base} + client, cleanup := newBufconnDistributionClient(t, flaky) + defer cleanup() + + mirror := distribution.NewEngineWithDefaultRoute() + var providerCalls atomic.Int32 + watcher := distribution.NewResolvingGRPCCatalogWatcher( + func(context.Context) (pb.DistributionClient, error) { + providerCalls.Add(1) + return client, nil + }, + mirror, + distribution.WithGRPCCatalogWatcherRetryInterval(time.Millisecond), + ) + watchCtx, cancel := context.WithCancel(context.Background()) + errCh := make(chan error, 1) + go func() { errCh <- watcher.Run(watchCtx) }() + + require.Eventually(t, func() bool { + return mirror.Version() == second.Version && flaky.watchCalls.Load() >= 2 && providerCalls.Load() >= 2 + }, 5*time.Second, 5*time.Millisecond) + cancel() + require.NoError(t, <-errCh) +} + +func TestGRPCCatalogWatcherReResolvesEndpointAfterDisconnect(t *testing.T) { + t.Parallel() + + ctx := context.Background() + catalog := distribution.NewCatalogStore(store.NewMVCCStore()) + routes := []distribution.RouteDescriptor{ + {RouteID: 1, Start: []byte(""), End: nil, GroupID: 1, State: distribution.RouteStateActive}, + } + first, err := catalog.Save(ctx, 0, routes) + require.NoError(t, err) + second, err := catalog.Save(ctx, first.Version, routes) + require.NoError(t, err) + + base := NewDistributionServer( + distribution.NewEngine(), + catalog, + WithCatalogWatchInterval(time.Millisecond), + ) + firstEndpoint := &disconnectAfterOneCatalogDeltaServer{DistributionServer: base} + firstClient, firstCleanup := newBufconnDistributionClient(t, firstEndpoint) + defer firstCleanup() + secondClient, secondCleanup := newBufconnDistributionClient(t, base) + defer secondCleanup() + + var resolutions atomic.Int32 + mirror := distribution.NewEngineWithDefaultRoute() + watcher := distribution.NewResolvingGRPCCatalogWatcher( + func(context.Context) (pb.DistributionClient, error) { + if resolutions.Add(1) == 1 { + return firstClient, nil + } + return secondClient, nil + }, + mirror, + distribution.WithGRPCCatalogWatcherRetryInterval(time.Millisecond), + ) + watchCtx, cancel := context.WithCancel(context.Background()) + errCh := make(chan error, 1) + go func() { errCh <- watcher.Run(watchCtx) }() + + require.Eventually(t, func() bool { + return mirror.Version() == second.Version && resolutions.Load() >= 2 + }, 5*time.Second, 5*time.Millisecond) + cancel() + require.NoError(t, <-errCh) +} + +func TestGRPCCatalogWatcherFallsBackToLegacyPolling(t *testing.T) { + t.Parallel() + + ctx := context.Background() + catalog := distribution.NewCatalogStore(store.NewMVCCStore()) + routes := []distribution.RouteDescriptor{ + {RouteID: 1, Start: []byte(""), End: nil, GroupID: 1, State: distribution.RouteStateActive}, + } + first, err := catalog.Save(ctx, 0, routes) + require.NoError(t, err) + + legacy := &legacyCatalogServer{catalog: catalog} + client, cleanup := newBufconnDistributionClient(t, legacy) + defer cleanup() + + mirror := distribution.NewEngineWithDefaultRoute() + watcher := distribution.NewGRPCCatalogWatcher( + client, + mirror, + distribution.WithGRPCCatalogWatcherRetryInterval(time.Millisecond), + ) + watchCtx, cancel := context.WithCancel(context.Background()) + errCh := make(chan error, 1) + go func() { errCh <- watcher.Run(watchCtx) }() + require.Eventually(t, func() bool { return mirror.Version() == first.Version }, 5*time.Second, 5*time.Millisecond) + + second, err := catalog.Save(ctx, first.Version, []distribution.RouteDescriptor{ + {RouteID: 2, Start: []byte(""), End: nil, GroupID: 2, State: distribution.RouteStateActive}, + }) + require.NoError(t, err) + require.Eventually(t, func() bool { return mirror.Version() == second.Version }, 5*time.Second, 5*time.Millisecond) + cancel() + require.NoError(t, <-errCh) +} + +func TestGRPCCatalogWatcherFallsBackWhenWatchRPCIsUnimplemented(t *testing.T) { + t.Parallel() + + ctx := context.Background() + catalog := distribution.NewCatalogStore(store.NewMVCCStore()) + routes := []distribution.RouteDescriptor{ + {RouteID: 1, Start: []byte(""), End: nil, GroupID: 1, State: distribution.RouteStateActive}, + } + first, err := catalog.Save(ctx, 0, routes) + require.NoError(t, err) + + partial := &capabilitiesWithoutWatchCatalogServer{catalog: catalog} + client, cleanup := newBufconnDistributionClient(t, partial) + defer cleanup() + + mirror := distribution.NewEngineWithDefaultRoute() + var watchLogs bytes.Buffer + watcher := distribution.NewGRPCCatalogWatcher( + client, + mirror, + distribution.WithGRPCCatalogWatcherRetryInterval(time.Millisecond), + distribution.WithGRPCCatalogWatcherLogger(slog.New(slog.NewTextHandler(&watchLogs, nil))), + ) + watchCtx, cancel := context.WithCancel(context.Background()) + errCh := make(chan error, 1) + go func() { errCh <- watcher.Run(watchCtx) }() + require.Eventually(t, func() bool { return mirror.Version() == first.Version }, 5*time.Second, 5*time.Millisecond) + time.Sleep(20 * time.Millisecond) + cancel() + require.NoError(t, <-errCh) + require.NotContains(t, watchLogs.String(), "catalog watch attempt failed") +} + +func TestCatalogStoreMutationsToOpsRejectsInvalidMutation(t *testing.T) { + t.Parallel() + + _, err := catalogStoreMutationsToOps([]*store.KVPairMutation{nil}) + require.ErrorIs(t, err, errDistributionCatalogMutationInvalid) + + _, err = catalogStoreMutationsToOps([]*store.KVPairMutation{{ + Op: store.OpType(99), + Key: []byte("invalid"), + }}) + require.ErrorIs(t, err, errDistributionCatalogMutationInvalid) +} + +type flakyCatalogWatchServer struct { + *DistributionServer + watchCalls atomic.Int32 +} + +type disconnectAfterOneCatalogDeltaServer struct { + *DistributionServer +} + +func (s *disconnectAfterOneCatalogDeltaServer) WatchCatalog( + req *pb.CatalogWatchRequest, + stream pb.Distribution_WatchCatalogServer, +) error { + changes, err := s.catalog.ChangesSince(stream.Context(), req.GetAfterVersion(), 1) + if err != nil { + return err + } + if len(changes.Deltas) != 1 { + return status.Error(codes.Internal, "injected disconnect requires one delta") + } + if err := stream.Send(&pb.CatalogWatchEvent{Payload: &pb.CatalogWatchEvent_Delta{ + Delta: toProtoCatalogDelta(changes.Deltas[0]), + }}); err != nil { + return err + } + return status.Error(codes.Unavailable, "injected endpoint replacement") +} + +func (s *flakyCatalogWatchServer) WatchCatalog(req *pb.CatalogWatchRequest, stream pb.Distribution_WatchCatalogServer) error { + if s.watchCalls.Add(1) != 1 { + return s.DistributionServer.WatchCatalog(req, stream) + } + changes, err := s.catalog.ChangesSince(stream.Context(), req.GetAfterVersion(), 1) + if err != nil { + return err + } + if len(changes.Deltas) != 1 { + return status.Error(codes.Internal, "injected disconnect requires one delta") + } + if err := stream.Send(&pb.CatalogWatchEvent{Payload: &pb.CatalogWatchEvent_Delta{ + Delta: toProtoCatalogDelta(changes.Deltas[0]), + }}); err != nil { + return err + } + return status.Error(codes.Unavailable, "injected catalog stream disconnect") +} + +type legacyCatalogServer struct { + pb.UnimplementedDistributionServer + catalog *distribution.CatalogStore +} + +type capabilitiesWithoutWatchCatalogServer struct { + pb.UnimplementedDistributionServer + catalog *distribution.CatalogStore +} + +func (s *capabilitiesWithoutWatchCatalogServer) GetCatalogCapabilities( + context.Context, + *pb.CatalogCapabilitiesRequest, +) (*pb.CatalogCapabilitiesResponse, error) { + return &pb.CatalogCapabilitiesResponse{ + SupportedProtocolVersions: []uint32{distribution.CatalogWatchProtocolVersion}, + MaxBatchSize: distribution.DefaultCatalogDeltaBatchSize, + }, nil +} + +func (s *capabilitiesWithoutWatchCatalogServer) ListRoutes( + ctx context.Context, + _ *pb.ListRoutesRequest, +) (*pb.ListRoutesResponse, error) { + snapshot, err := s.catalog.Snapshot(ctx) + if err != nil { + return nil, err + } + return &pb.ListRoutesResponse{ + CatalogVersion: snapshot.Version, + Routes: toProtoRouteDescriptors(snapshot.Routes), + }, nil +} + +func (s *legacyCatalogServer) ListRoutes(ctx context.Context, _ *pb.ListRoutesRequest) (*pb.ListRoutesResponse, error) { + snapshot, err := s.catalog.Snapshot(ctx) + if err != nil { + return nil, err + } + return &pb.ListRoutesResponse{ + CatalogVersion: snapshot.Version, + Routes: toProtoRouteDescriptors(snapshot.Routes), + }, nil +} + +func newBufconnDistributionClient(t *testing.T, server pb.DistributionServer) (pb.DistributionClient, func()) { + t.Helper() + + listener := bufconn.Listen(1024 * 1024) + grpcServer := grpc.NewServer() + pb.RegisterDistributionServer(grpcServer, server) + serveErr := make(chan error, 1) + go func() { serveErr <- grpcServer.Serve(listener) }() + + conn, err := grpc.NewClient( + "passthrough:///bufnet", + grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) { + return listener.Dial() + }), + grpc.WithTransportCredentials(insecure.NewCredentials()), + ) + require.NoError(t, err) + + cleanup := func() { + require.NoError(t, conn.Close()) + grpcServer.Stop() + err := <-serveErr + if err != nil { + require.ErrorIs(t, err, grpc.ErrServerStopped) + } + require.NoError(t, listener.Close()) + } + return pb.NewDistributionClient(conn), cleanup +} + +type captureCatalogWatchStream struct { + ctx context.Context + cancel context.CancelFunc + events chan *pb.CatalogWatchEvent + limit int +} + +func newCaptureCatalogWatchStream() *captureCatalogWatchStream { + const limit = 1 + ctx, cancel := context.WithCancel(context.Background()) + return &captureCatalogWatchStream{ + ctx: ctx, + cancel: cancel, + events: make(chan *pb.CatalogWatchEvent, limit), + limit: limit, + } +} + +func (s *captureCatalogWatchStream) Send(event *pb.CatalogWatchEvent) error { + s.events <- event + if len(s.events) >= s.limit { + s.cancel() + } + return nil +} + +func (s *captureCatalogWatchStream) SetHeader(metadata.MD) error { return nil } +func (s *captureCatalogWatchStream) SendHeader(metadata.MD) error { return nil } +func (s *captureCatalogWatchStream) SetTrailer(metadata.MD) {} +func (s *captureCatalogWatchStream) Context() context.Context { return s.ctx } +func (s *captureCatalogWatchStream) SendMsg(message any) error { + event, ok := message.(*pb.CatalogWatchEvent) + if !ok { + return io.ErrUnexpectedEOF + } + return s.Send(event) +} +func (s *captureCatalogWatchStream) RecvMsg(any) error { return io.EOF } diff --git a/adapter/distribution_server_test.go b/adapter/distribution_server_test.go index e645d4d96..ce268cddc 100644 --- a/adapter/distribution_server_test.go +++ b/adapter/distribution_server_test.go @@ -499,8 +499,9 @@ func TestDistributionServerSplitRange_UsesCoordinatorForCatalogWrites(t *testing require.Equal(t, uint64(2), resp.CatalogVersion) require.Equal(t, 1, coordinator.dispatchCalls) require.Equal(t, readSnapshot.ReadTS, coordinator.lastStartTS) - require.Zero(t, coordinator.lastRequestedCommitTS) + require.NotZero(t, coordinator.lastRequestedCommitTS) require.NotZero(t, coordinator.lastCommitTS) + require.Equal(t, coordinator.lastRequestedCommitTS, coordinator.lastCommitTS) require.Greater(t, coordinator.lastCommitTS, coordinator.lastStartTS) snapshot, err := catalog.Snapshot(ctx) @@ -513,6 +514,14 @@ func TestDistributionServerSplitRange_UsesCoordinatorForCatalogWrites(t *testing require.Equal(t, coordinator.lastCommitTS, right.SplitAtHLC) require.Equal(t, coordinator.lastCommitTS, resp.Left.SplitAtHlc) require.Equal(t, coordinator.lastCommitTS, resp.Right.SplitAtHlc) + + changes, err := catalog.ChangesSince(ctx, saved.Version, 1) + require.NoError(t, err) + require.Len(t, changes.Deltas, 1) + require.Len(t, changes.Deltas[0].Mutations, 3) + require.Equal(t, distribution.CatalogMutationDelete, changes.Deltas[0].Mutations[0].Op) + require.Equal(t, coordinator.lastCommitTS, changes.Deltas[0].Mutations[1].Route.SplitAtHLC) + require.Equal(t, coordinator.lastCommitTS, changes.Deltas[0].Mutations[2].Route.SplitAtHLC) } func TestDistributionServerSplitRange_UsesPersistentNextRouteID(t *testing.T) { diff --git a/distribution/catalog.go b/distribution/catalog.go index b92f85f23..a27734c1f 100644 --- a/distribution/catalog.go +++ b/distribution/catalog.go @@ -630,6 +630,15 @@ func (s *CatalogStore) buildSaveMutations(ctx context.Context, plan savePlan) ([ if err != nil { return nil, err } + delta, err := buildCatalogDelta(existingRoutes, plan.routes, plan.nextVersion-1, plan.nextVersion) + if err != nil { + return nil, err + } + deltaMutations, err := s.BuildDeltaMutationsAt(ctx, plan.readTS, delta) + if err != nil { + return nil, err + } + mutations = append(mutations, deltaMutations...) mutations = append(mutations, &store.KVPairMutation{ Op: store.OpTypePut, Key: CatalogVersionKey(), diff --git a/distribution/catalog_delta.go b/distribution/catalog_delta.go new file mode 100644 index 000000000..c051ebd53 --- /dev/null +++ b/distribution/catalog_delta.go @@ -0,0 +1,483 @@ +package distribution + +import ( + "bytes" + "context" + "encoding/binary" + "math" + "sort" + + "github.com/bootjp/elastickv/store" + "github.com/cockroachdb/errors" +) + +const ( + catalogDeltaPrefix = "!dist|delta|" + catalogDeltaFloorStorageKey = catalogMetaPrefix + "delta_floor" + + catalogDeltaCodecVersion byte = 1 + catalogDeltaHeaderSize = 1 + 3*catalogUint64Bytes + catalogDeltaMutationHeaderSize = 1 + 2*catalogUint64Bytes + + // DefaultCatalogDeltaRetention bounds the durable reconnect window. A + // watcher that falls behind this many catalog versions receives a snapshot + // reset instead of an incomplete delta sequence. + DefaultCatalogDeltaRetention uint64 = 1024 + DefaultCatalogDeltaBatchSize = 128 + MaxCatalogDeltaBatchSize = 1024 + CatalogWatchProtocolVersion uint32 = 1 +) + +var ( + ErrCatalogInvalidDeltaRecord = errors.New("catalog delta record is invalid") + ErrCatalogInvalidDeltaMutation = errors.New("catalog delta mutation is invalid") + ErrCatalogDeltaVersionGap = errors.New("catalog delta version is not contiguous") + ErrCatalogDeltaBaseMismatch = errors.New("catalog delta base version does not match durable catalog") + ErrCatalogDeltaVersionFuture = errors.New("catalog delta cursor is ahead of catalog version") + ErrCatalogDeltaLimitInvalid = errors.New("catalog delta limit must be positive") +) + +// CatalogMutationOp describes one route-level catalog change. +type CatalogMutationOp byte + +const ( + CatalogMutationUpsert CatalogMutationOp = 1 + CatalogMutationDelete CatalogMutationOp = 2 +) + +// CatalogRouteMutation changes one durable route descriptor. +type CatalogRouteMutation struct { + Op CatalogMutationOp + RouteID uint64 + Route RouteDescriptor +} + +// CatalogDelta is the durable transition from PreviousVersion to Version. +type CatalogDelta struct { + PreviousVersion uint64 + Version uint64 + Mutations []CatalogRouteMutation +} + +// CatalogChangeSet contains either a snapshot reset or a contiguous delta +// batch. Reset is non-nil only when the requested cursor predates retention. +type CatalogChangeSet struct { + Reset *CatalogSnapshot + Deltas []CatalogDelta +} + +// CatalogDeltaFloorKey stores the oldest retained delta version. +func CatalogDeltaFloorKey() []byte { + return []byte(catalogDeltaFloorStorageKey) +} + +// CatalogDeltaKey returns the ordered durable key for a catalog delta. +func CatalogDeltaKey(version uint64) []byte { + key := make([]byte, len(catalogDeltaPrefix)+catalogUint64Bytes) + copy(key, catalogDeltaPrefix) + binary.BigEndian.PutUint64(key[len(catalogDeltaPrefix):], version) + return key +} + +// IsCatalogDeltaKey reports whether key belongs to the delta log keyspace. +func IsCatalogDeltaKey(key []byte) bool { + return bytes.HasPrefix(key, []byte(catalogDeltaPrefix)) +} + +// CatalogDeltaVersionFromKey parses the version from a durable delta key. +func CatalogDeltaVersionFromKey(key []byte) (uint64, bool) { + if !IsCatalogDeltaKey(key) { + return 0, false + } + suffix := key[len(catalogDeltaPrefix):] + if len(suffix) != catalogUint64Bytes { + return 0, false + } + version := binary.BigEndian.Uint64(suffix) + return version, version != 0 +} + +// EncodeCatalogDelta serializes a validated catalog transition. +func EncodeCatalogDelta(delta CatalogDelta) ([]byte, error) { + if err := validateCatalogDelta(delta); err != nil { + return nil, err + } + if len(delta.Mutations) > (math.MaxInt-catalogDeltaHeaderSize)/catalogDeltaMutationHeaderSize { + return nil, errors.WithStack(ErrCatalogInvalidDeltaRecord) + } + + out := make([]byte, 0, catalogDeltaHeaderSize+len(delta.Mutations)*catalogDeltaMutationHeaderSize) + out = append(out, catalogDeltaCodecVersion) + out = appendU64(out, delta.PreviousVersion) + out = appendU64(out, delta.Version) + out = appendU64(out, uint64(len(delta.Mutations))) + for _, mutation := range delta.Mutations { + out = append(out, byte(mutation.Op)) + out = appendU64(out, mutation.RouteID) + if mutation.Op == CatalogMutationDelete { + out = appendU64(out, 0) + continue + } + encoded, err := EncodeRouteDescriptor(mutation.Route) + if err != nil { + return nil, err + } + out = appendU64(out, uint64(len(encoded))) + out = append(out, encoded...) + } + return out, nil +} + +// DecodeCatalogDelta deserializes and validates a catalog transition. +func DecodeCatalogDelta(raw []byte) (CatalogDelta, error) { + if len(raw) < catalogDeltaHeaderSize || raw[0] != catalogDeltaCodecVersion { + return CatalogDelta{}, errors.WithStack(ErrCatalogInvalidDeltaRecord) + } + r := bytes.NewReader(raw[1:]) + delta, count, err := decodeCatalogDeltaHeader(r) + if err != nil { + return CatalogDelta{}, err + } + delta.Mutations, err = decodeCatalogDeltaMutations(r, count) + if err != nil { + return CatalogDelta{}, err + } + if r.Len() != 0 { + return CatalogDelta{}, errors.WithStack(ErrCatalogInvalidDeltaRecord) + } + if err := validateCatalogDelta(delta); err != nil { + return CatalogDelta{}, err + } + return delta, nil +} + +func decodeCatalogDeltaHeader(r *bytes.Reader) (CatalogDelta, int, error) { + var delta CatalogDelta + var mutationCount uint64 + fields := []*uint64{&delta.PreviousVersion, &delta.Version, &mutationCount} + for _, field := range fields { + if err := binary.Read(r, binary.BigEndian, field); err != nil { + return CatalogDelta{}, 0, errors.WithStack(ErrCatalogInvalidDeltaRecord) + } + } + count, err := u64ToInt(mutationCount) + if err != nil || count > r.Len()/catalogDeltaMutationHeaderSize { + return CatalogDelta{}, 0, errors.WithStack(ErrCatalogInvalidDeltaRecord) + } + return delta, count, nil +} + +func decodeCatalogDeltaMutations(r *bytes.Reader, count int) ([]CatalogRouteMutation, error) { + mutations := make([]CatalogRouteMutation, 0, count) + for range count { + mutation, err := decodeCatalogRouteMutation(r) + if err != nil { + return nil, err + } + mutations = append(mutations, mutation) + } + return mutations, nil +} + +func decodeCatalogRouteMutation(r *bytes.Reader) (CatalogRouteMutation, error) { + op, err := r.ReadByte() + if err != nil { + return CatalogRouteMutation{}, errors.WithStack(ErrCatalogInvalidDeltaRecord) + } + mutation := CatalogRouteMutation{Op: CatalogMutationOp(op)} + var payloadLen uint64 + if err := binary.Read(r, binary.BigEndian, &mutation.RouteID); err != nil { + return CatalogRouteMutation{}, errors.WithStack(ErrCatalogInvalidDeltaRecord) + } + if err := binary.Read(r, binary.BigEndian, &payloadLen); err != nil { + return CatalogRouteMutation{}, errors.WithStack(ErrCatalogInvalidDeltaRecord) + } + payload, err := readU64LenBytes(r, payloadLen) + if err != nil { + return CatalogRouteMutation{}, errors.WithStack(ErrCatalogInvalidDeltaRecord) + } + if mutation.Op == CatalogMutationUpsert { + mutation.Route, err = DecodeRouteDescriptor(payload) + if err != nil { + return CatalogRouteMutation{}, err + } + } else if len(payload) != 0 { + return CatalogRouteMutation{}, errors.WithStack(ErrCatalogInvalidDeltaMutation) + } + return mutation, nil +} + +func validateCatalogDelta(delta CatalogDelta) error { + if delta.Version == 0 || delta.PreviousVersion == math.MaxUint64 || delta.PreviousVersion+1 != delta.Version { + return errors.WithStack(ErrCatalogDeltaVersionGap) + } + seen := make(map[uint64]struct{}, len(delta.Mutations)) + for _, mutation := range delta.Mutations { + if mutation.RouteID == 0 { + return errors.WithStack(ErrCatalogInvalidDeltaMutation) + } + if _, ok := seen[mutation.RouteID]; ok { + return errors.WithStack(ErrCatalogInvalidDeltaMutation) + } + seen[mutation.RouteID] = struct{}{} + if err := validateCatalogDeltaMutation(mutation); err != nil { + return err + } + } + return nil +} + +func validateCatalogDeltaMutation(mutation CatalogRouteMutation) error { + switch mutation.Op { + case CatalogMutationUpsert: + if mutation.Route.RouteID != mutation.RouteID { + return errors.WithStack(ErrCatalogInvalidDeltaMutation) + } + return validateRouteDescriptor(mutation.Route) + case CatalogMutationDelete: + if !catalogDeltaRouteIsZero(mutation.Route) { + return errors.WithStack(ErrCatalogInvalidDeltaMutation) + } + return nil + default: + return errors.WithStack(ErrCatalogInvalidDeltaMutation) + } +} + +func catalogDeltaRouteIsZero(route RouteDescriptor) bool { + return route.RouteID == 0 && + route.Start == nil && + route.End == nil && + route.GroupID == 0 && + route.State == 0 && + route.ParentRouteID == 0 && + route.SplitAtHLC == 0 +} + +func buildCatalogDelta(existing, desired []RouteDescriptor, previousVersion, version uint64) (CatalogDelta, error) { + existingByID := make(map[uint64]RouteDescriptor, len(existing)) + desiredByID := make(map[uint64]RouteDescriptor, len(desired)) + for _, route := range existing { + existingByID[route.RouteID] = route + } + for _, route := range desired { + desiredByID[route.RouteID] = route + } + + mutations := make([]CatalogRouteMutation, 0) + for routeID := range existingByID { + if _, ok := desiredByID[routeID]; !ok { + mutations = append(mutations, CatalogRouteMutation{Op: CatalogMutationDelete, RouteID: routeID}) + } + } + for routeID, route := range desiredByID { + if current, ok := existingByID[routeID]; ok && routeDescriptorEqual(current, route) { + continue + } + mutations = append(mutations, CatalogRouteMutation{ + Op: CatalogMutationUpsert, + RouteID: routeID, + Route: CloneRouteDescriptor(route), + }) + } + sort.Slice(mutations, func(i, j int) bool { return mutations[i].RouteID < mutations[j].RouteID }) + delta := CatalogDelta{PreviousVersion: previousVersion, Version: version, Mutations: mutations} + return delta, validateCatalogDelta(delta) +} + +// BuildDeltaMutationsAt returns the durable delta/floor/retention mutations +// that must be committed atomically with the catalog version transition. +func (s *CatalogStore) BuildDeltaMutationsAt(ctx context.Context, readTS uint64, delta CatalogDelta) ([]*store.KVPairMutation, error) { + if err := ensureCatalogStore(s); err != nil { + return nil, err + } + ctx = contextOrBackground(ctx) + if err := validateCatalogDelta(delta); err != nil { + return nil, err + } + currentVersion, err := s.versionAt(ctx, readTS) + if err != nil { + return nil, err + } + if currentVersion != delta.PreviousVersion { + return nil, errors.Wrapf( + ErrCatalogDeltaBaseMismatch, + "delta follows %d, durable catalog is at %d", + delta.PreviousVersion, + currentVersion, + ) + } + encoded, err := EncodeCatalogDelta(delta) + if err != nil { + return nil, err + } + currentFloor, err := s.deltaFloorAt(ctx, readTS) + if err != nil { + return nil, err + } + newFloor := nextCatalogDeltaFloor(currentFloor, delta.PreviousVersion, delta.Version) + mutations := []*store.KVPairMutation{ + {Op: store.OpTypePut, Key: CatalogDeltaKey(delta.Version), Value: encoded}, + {Op: store.OpTypePut, Key: CatalogDeltaFloorKey(), Value: EncodeCatalogVersion(newFloor)}, + } + if currentFloor != 0 && newFloor > currentFloor { + for version := currentFloor; version < newFloor; version++ { + mutations = append(mutations, &store.KVPairMutation{Op: store.OpTypeDelete, Key: CatalogDeltaKey(version)}) + } + } + return mutations, nil +} + +func nextCatalogDeltaFloor(currentFloor, previousVersion, version uint64) uint64 { + floor := currentFloor + if floor == 0 { + floor = 1 + if previousVersion != 0 { + floor = version + } + } + if version >= DefaultCatalogDeltaRetention { + retentionFloor := version - DefaultCatalogDeltaRetention + 1 + if retentionFloor > floor { + floor = retentionFloor + } + } + return floor +} + +func (s *CatalogStore) deltaFloorAt(ctx context.Context, ts uint64) (uint64, error) { + raw, err := s.store.GetAt(ctx, CatalogDeltaFloorKey(), ts) + if err != nil { + if errors.Is(err, store.ErrKeyNotFound) { + return 0, nil + } + return 0, errors.WithStack(err) + } + floor, err := DecodeCatalogVersion(raw) + if err != nil || floor == 0 { + return 0, errors.WithStack(ErrCatalogInvalidDeltaRecord) + } + return floor, nil +} + +// DeltaFloor returns the oldest retained durable delta version. Zero means the +// catalog predates delta logging and a watcher must request a snapshot reset. +func (s *CatalogStore) DeltaFloor(ctx context.Context) (uint64, error) { + if err := ensureCatalogStore(s); err != nil { + return 0, err + } + ctx = contextOrBackground(ctx) + return s.deltaFloorAt(ctx, s.store.LastCommitTS()) +} + +// ChangesSince reads a consistent reconnect batch after afterVersion. +func (s *CatalogStore) ChangesSince(ctx context.Context, afterVersion uint64, limit int) (CatalogChangeSet, error) { + if err := ensureCatalogStore(s); err != nil { + return CatalogChangeSet{}, err + } + if limit <= 0 { + return CatalogChangeSet{}, errors.WithStack(ErrCatalogDeltaLimitInvalid) + } + ctx = contextOrBackground(ctx) + readTS := s.store.LastCommitTS() + currentVersion, err := s.versionAt(ctx, readTS) + if err != nil { + return CatalogChangeSet{}, err + } + requested, current, err := nextCatalogChangeVersion(afterVersion, currentVersion) + if err != nil { + return CatalogChangeSet{}, err + } + if current { + return CatalogChangeSet{}, nil + } + floor, err := s.deltaFloorAt(ctx, readTS) + if err != nil { + return CatalogChangeSet{}, err + } + if floor > currentVersion { + return CatalogChangeSet{}, errors.Wrapf( + ErrCatalogInvalidDeltaRecord, + "delta floor %d exceeds catalog version %d", + floor, + currentVersion, + ) + } + if floor == 0 || requested < floor { + return s.catalogSnapshotResetAt(ctx, readTS) + } + return s.catalogDeltasAt(ctx, readTS, requested, currentVersion, limit) +} + +func nextCatalogChangeVersion(afterVersion, currentVersion uint64) (uint64, bool, error) { + if afterVersion > currentVersion { + return 0, false, errors.WithStack(ErrCatalogDeltaVersionFuture) + } + if afterVersion == currentVersion { + return 0, true, nil + } + if afterVersion == math.MaxUint64 { + return 0, false, errors.WithStack(ErrCatalogVersionOverflow) + } + return afterVersion + 1, false, nil +} + +func (s *CatalogStore) catalogSnapshotResetAt(ctx context.Context, readTS uint64) (CatalogChangeSet, error) { + snapshot, err := s.snapshotAt(ctx, readTS) + if err != nil { + return CatalogChangeSet{}, err + } + return CatalogChangeSet{Reset: &snapshot}, nil +} + +func (s *CatalogStore) catalogDeltasAt( + ctx context.Context, + readTS uint64, + requested uint64, + currentVersion uint64, + limit int, +) (CatalogChangeSet, error) { + deltas := make([]CatalogDelta, 0) + for version := requested; version <= currentVersion && len(deltas) < limit; version++ { + delta, err := s.catalogDeltaAt(ctx, readTS, version) + if err != nil { + return CatalogChangeSet{}, err + } + deltas = append(deltas, delta) + if version == math.MaxUint64 { + break + } + } + return CatalogChangeSet{Deltas: deltas}, nil +} + +func (s *CatalogStore) catalogDeltaAt(ctx context.Context, readTS, version uint64) (CatalogDelta, error) { + raw, err := s.store.GetAt(ctx, CatalogDeltaKey(version), readTS) + if err != nil { + if errors.Is(err, store.ErrKeyNotFound) { + return CatalogDelta{}, errors.Wrapf(ErrCatalogDeltaVersionGap, "missing version %d", version) + } + return CatalogDelta{}, errors.WithStack(err) + } + delta, err := DecodeCatalogDelta(raw) + if err != nil { + return CatalogDelta{}, err + } + expectedPrevious := version - 1 + if delta.Version != version || delta.PreviousVersion != expectedPrevious { + return CatalogDelta{}, errors.Wrapf(ErrCatalogDeltaVersionGap, "got %d after %d", delta.Version, expectedPrevious) + } + return delta, nil +} + +func (s *CatalogStore) snapshotAt(ctx context.Context, readTS uint64) (CatalogSnapshot, error) { + version, err := s.versionAt(ctx, readTS) + if err != nil { + return CatalogSnapshot{}, err + } + routes, err := s.routesAt(ctx, readTS) + if err != nil { + return CatalogSnapshot{}, err + } + return CatalogSnapshot{Version: version, Routes: routes, ReadTS: readTS}, nil +} diff --git a/distribution/catalog_delta_test.go b/distribution/catalog_delta_test.go new file mode 100644 index 000000000..ca47988c8 --- /dev/null +++ b/distribution/catalog_delta_test.go @@ -0,0 +1,257 @@ +package distribution + +import ( + "context" + "testing" + + pb "github.com/bootjp/elastickv/proto" + "github.com/bootjp/elastickv/store" + "github.com/stretchr/testify/require" +) + +func TestCatalogDeltaCodecRoundTrip(t *testing.T) { + t.Parallel() + + delta := CatalogDelta{ + PreviousVersion: 7, + Version: 8, + Mutations: []CatalogRouteMutation{ + {Op: CatalogMutationDelete, RouteID: 1}, + { + Op: CatalogMutationUpsert, + RouteID: 2, + Route: RouteDescriptor{ + RouteID: 2, Start: []byte("m"), GroupID: 3, + State: RouteStateMigratingTarget, ParentRouteID: 1, + }, + }, + }, + } + + raw, err := EncodeCatalogDelta(delta) + require.NoError(t, err) + decoded, err := DecodeCatalogDelta(raw) + require.NoError(t, err) + require.Equal(t, delta, decoded) +} + +func TestCatalogDeltaRejectsDeleteRoutePayload(t *testing.T) { + t.Parallel() + + for _, route := range []RouteDescriptor{ + {Start: []byte("unexpected")}, + {Start: []byte{}}, + } { + _, err := EncodeCatalogDelta(CatalogDelta{ + PreviousVersion: 1, + Version: 2, + Mutations: []CatalogRouteMutation{{ + Op: CatalogMutationDelete, + RouteID: 1, + Route: route, + }}, + }) + require.ErrorIs(t, err, ErrCatalogInvalidDeltaMutation) + } +} + +func TestCatalogDeltaFromProtoRejectsDeleteRoutePayload(t *testing.T) { + t.Parallel() + + _, err := catalogDeltaFromProto(&pb.CatalogDeltaRecord{ + PreviousVersion: 1, + Version: 2, + Mutations: []*pb.CatalogDeltaMutation{{ + Op: pb.CatalogDeltaMutationOp_CATALOG_DELTA_MUTATION_OP_DELETE, + RouteId: 1, + Route: &pb.RouteDescriptor{}, + }}, + }) + require.ErrorIs(t, err, ErrCatalogWatchEventInvalid) +} + +func TestCatalogStoreSavePublishesContiguousDeltas(t *testing.T) { + t.Parallel() + + ctx := context.Background() + st := store.NewMVCCStore() + catalog := NewCatalogStore(st) + firstRoutes := []RouteDescriptor{ + {RouteID: 1, Start: []byte(""), End: nil, GroupID: 1, State: RouteStateActive}, + } + first, err := catalog.Save(ctx, 0, firstRoutes) + require.NoError(t, err) + secondRoutes := []RouteDescriptor{ + {RouteID: 2, Start: []byte(""), End: []byte("m"), GroupID: 1, State: RouteStateActive, ParentRouteID: 1}, + {RouteID: 3, Start: []byte("m"), End: nil, GroupID: 2, State: RouteStateActive, ParentRouteID: 1}, + } + second, err := catalog.Save(ctx, first.Version, secondRoutes) + require.NoError(t, err) + + floor, err := catalog.DeltaFloor(ctx) + require.NoError(t, err) + require.EqualValues(t, 1, floor) + changes, err := catalog.ChangesSince(ctx, 0, 10) + require.NoError(t, err) + require.Nil(t, changes.Reset) + require.Len(t, changes.Deltas, 2) + require.EqualValues(t, 0, changes.Deltas[0].PreviousVersion) + require.EqualValues(t, 1, changes.Deltas[0].Version) + require.EqualValues(t, 1, changes.Deltas[1].PreviousVersion) + require.EqualValues(t, 2, changes.Deltas[1].Version) + + engine := NewEngineWithDefaultRoute() + for _, delta := range changes.Deltas { + require.NoError(t, engine.ApplyDelta(delta)) + } + require.Equal(t, second.Version, engine.Version()) + routes := engine.Stats() + require.Len(t, routes, 2) + require.EqualValues(t, 2, routes[0].RouteID) + require.EqualValues(t, 3, routes[1].RouteID) +} + +func TestCatalogStoreChangesSinceFallsBackAfterRetention(t *testing.T) { + t.Parallel() + + ctx := context.Background() + st := store.NewMVCCStore() + catalog := NewCatalogStore(st) + routes := []RouteDescriptor{ + {RouteID: 1, Start: []byte(""), End: nil, GroupID: 1, State: RouteStateActive}, + } + version := uint64(0) + for range DefaultCatalogDeltaRetention + 1 { + snapshot, err := catalog.Save(ctx, version, routes) + require.NoError(t, err) + version = snapshot.Version + } + + floor, err := catalog.DeltaFloor(ctx) + require.NoError(t, err) + require.EqualValues(t, 2, floor) + _, err = st.GetAt(ctx, CatalogDeltaKey(1), st.LastCommitTS()) + require.ErrorIs(t, err, store.ErrKeyNotFound) + + changes, err := catalog.ChangesSince(ctx, 0, 10) + require.NoError(t, err) + require.NotNil(t, changes.Reset) + require.Equal(t, version, changes.Reset.Version) + require.Equal(t, routes, changes.Reset.Routes) + require.Empty(t, changes.Deltas) +} + +func TestCatalogStoreChangesSinceRejectsFutureCursor(t *testing.T) { + t.Parallel() + + catalog := NewCatalogStore(store.NewMVCCStore()) + _, err := catalog.ChangesSince(context.Background(), 1, 1) + require.ErrorIs(t, err, ErrCatalogDeltaVersionFuture) +} + +func TestCatalogStoreBuildDeltaRejectsDurableBaseMismatch(t *testing.T) { + t.Parallel() + + catalog := NewCatalogStore(store.NewMVCCStore()) + _, err := catalog.BuildDeltaMutationsAt(context.Background(), 0, CatalogDelta{ + PreviousVersion: 1, + Version: 2, + }) + require.ErrorIs(t, err, ErrCatalogDeltaBaseMismatch) +} + +func TestCatalogStoreChangesSinceRejectsMissingRetainedDelta(t *testing.T) { + t.Parallel() + + ctx := context.Background() + st := store.NewMVCCStore() + catalog := NewCatalogStore(st) + routes := []RouteDescriptor{ + {RouteID: 1, Start: []byte(""), End: nil, GroupID: 1, State: RouteStateActive}, + } + first, err := catalog.Save(ctx, 0, routes) + require.NoError(t, err) + second, err := catalog.Save(ctx, first.Version, routes) + require.NoError(t, err) + startTS := st.LastCommitTS() + require.NoError(t, st.ApplyMutations(ctx, []*store.KVPairMutation{ + {Op: store.OpTypeDelete, Key: CatalogDeltaKey(second.Version)}, + }, nil, startTS, startTS+1)) + + _, err = catalog.ChangesSince(ctx, first.Version, 1) + require.ErrorIs(t, err, ErrCatalogDeltaVersionGap) +} + +func TestEngineApplyDeltaRejectsGapWithoutPublishing(t *testing.T) { + t.Parallel() + + engine := NewEngine() + require.NoError(t, engine.ApplySnapshot(CatalogSnapshot{ + Version: 3, + Routes: []RouteDescriptor{ + {RouteID: 1, Start: []byte(""), End: nil, GroupID: 1, State: RouteStateActive}, + }, + })) + before := engine.Stats() + err := engine.ApplyDelta(CatalogDelta{ + PreviousVersion: 2, + Version: 3, + }) + require.NoError(t, err, "replaying the published version is idempotent") + err = engine.ApplyDelta(CatalogDelta{ + PreviousVersion: 4, + Version: 5, + }) + require.ErrorIs(t, err, ErrEngineDeltaVersionGap) + require.EqualValues(t, 3, engine.Version()) + require.Equal(t, before, engine.Stats()) +} + +func TestEngineApplyDeltaRejectsOverlapWithoutPublishing(t *testing.T) { + t.Parallel() + + engine := NewEngine() + require.NoError(t, engine.ApplySnapshot(CatalogSnapshot{ + Version: 1, + Routes: []RouteDescriptor{ + {RouteID: 1, Start: []byte(""), End: []byte("m"), GroupID: 1, State: RouteStateActive}, + {RouteID: 2, Start: []byte("m"), End: nil, GroupID: 2, State: RouteStateActive}, + }, + })) + before := engine.Stats() + err := engine.ApplyDelta(CatalogDelta{ + PreviousVersion: 1, + Version: 2, + Mutations: []CatalogRouteMutation{{ + Op: CatalogMutationUpsert, + RouteID: 2, + Route: RouteDescriptor{ + RouteID: 2, Start: []byte("l"), End: nil, GroupID: 2, State: RouteStateActive, + }, + }}, + }) + require.ErrorIs(t, err, ErrEngineSnapshotRouteOverlap) + require.EqualValues(t, 1, engine.Version()) + require.Equal(t, before, engine.Stats()) +} + +func TestCatalogWatcherAppliesBoundedDeltaBatches(t *testing.T) { + t.Parallel() + + ctx := context.Background() + catalog := NewCatalogStore(store.NewMVCCStore()) + routes := []RouteDescriptor{ + {RouteID: 1, Start: []byte(""), End: nil, GroupID: 1, State: RouteStateActive}, + } + first, err := catalog.Save(ctx, 0, routes) + require.NoError(t, err) + second, err := catalog.Save(ctx, first.Version, routes) + require.NoError(t, err) + + engine := NewEngineWithDefaultRoute() + watcher := NewCatalogWatcher(catalog, engine, WithCatalogWatcherBatchSize(1)) + require.NoError(t, watcher.SyncOnce(ctx)) + require.Equal(t, first.Version, engine.Version()) + require.NoError(t, watcher.SyncOnce(ctx)) + require.Equal(t, second.Version, engine.Version()) +} diff --git a/distribution/engine.go b/distribution/engine.go index bc6613894..3af787314 100644 --- a/distribution/engine.go +++ b/distribution/engine.go @@ -64,6 +64,7 @@ var ( ErrEngineSnapshotDuplicateID = errors.New("engine snapshot has duplicate route id") ErrEngineSnapshotRouteOverlap = errors.New("engine snapshot has overlapping routes") ErrEngineSnapshotRouteOrder = errors.New("engine snapshot has invalid route order") + ErrEngineDeltaVersionGap = errors.New("engine catalog delta version is not contiguous") ) // NewEngine creates an Engine with no hotspot splitting. @@ -130,6 +131,92 @@ func (e *Engine) ApplySnapshot(snapshot CatalogSnapshot) error { return nil } +// ApplyDelta atomically publishes one contiguous catalog transition. Readers +// observe either the complete previous route table or the complete next table. +func (e *Engine) ApplyDelta(delta CatalogDelta) error { + if err := validateCatalogDelta(delta); err != nil { + return err + } + + e.mu.Lock() + defer e.mu.Unlock() + apply, err := validateEngineDeltaVersion(delta, e.catalogVersion) + if err != nil { + return err + } + if !apply { + return nil + } + next, err := routesAfterCatalogDelta(e.routes, delta) + if err != nil { + return err + } + e.routes = next + e.catalogVersion = delta.Version + e.recordHistorySnapshotLocked() + return nil +} + +func validateEngineDeltaVersion(delta CatalogDelta, currentVersion uint64) (bool, error) { + if delta.Version < currentVersion { + return false, staleSnapshotVersionErr(delta.Version, currentVersion) + } + if delta.Version == currentVersion { + return false, nil + } + if delta.PreviousVersion != currentVersion { + return false, errors.Wrapf( + ErrEngineDeltaVersionGap, + "delta %d follows %d, engine is at %d", + delta.Version, + delta.PreviousVersion, + currentVersion, + ) + } + return true, nil +} + +func routesAfterCatalogDelta(current []Route, delta CatalogDelta) ([]Route, error) { + byID := make(map[uint64]Route, len(current)+len(delta.Mutations)) + for _, route := range current { + if delta.PreviousVersion == 0 && route.RouteID == 0 { + continue + } + byID[route.RouteID] = route + } + for _, mutation := range delta.Mutations { + switch mutation.Op { + case CatalogMutationDelete: + delete(byID, mutation.RouteID) + case CatalogMutationUpsert: + load := uint64(0) + if current, ok := byID[mutation.RouteID]; ok { + load = current.Load + } + byID[mutation.RouteID] = Route{ + RouteID: mutation.Route.RouteID, + Start: CloneBytes(mutation.Route.Start), + End: CloneBytes(mutation.Route.End), + GroupID: mutation.Route.GroupID, + State: mutation.Route.State, + Load: load, + } + } + } + + next := make([]Route, 0, len(byID)) + for _, route := range byID { + next = append(next, route) + } + sort.Slice(next, func(i, j int) bool { + return bytes.Compare(next[i].Start, next[j].Start) < 0 + }) + if err := validateRouteOrder(next); err != nil { + return nil, err + } + return next, nil +} + // RouteHistorySnapshot is a point-in-time view of the route catalog at // a specific version. Returned by Engine.SnapshotAt for the M3 // Composed-1 commit-time gate. Carries an immutable copy of the diff --git a/distribution/grpc_watcher.go b/distribution/grpc_watcher.go new file mode 100644 index 000000000..96ce2faf5 --- /dev/null +++ b/distribution/grpc_watcher.go @@ -0,0 +1,353 @@ +package distribution + +import ( + "context" + "io" + "log/slog" + "time" + + pb "github.com/bootjp/elastickv/proto" + "github.com/cockroachdb/errors" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +const defaultGRPCCatalogWatcherRetryInterval = 250 * time.Millisecond + +var ( + ErrCatalogWatchClientRequired = errors.New("catalog watch client is required") + ErrCatalogWatchEventInvalid = errors.New("catalog watch event is invalid") +) + +// CatalogWatchClientProvider resolves the Distribution endpoint used for one +// capability negotiation and stream attempt. Production resolves the current +// default-group leader again after every disconnect. +type CatalogWatchClientProvider func(context.Context) (pb.DistributionClient, error) + +// GRPCCatalogWatcherOption customizes a remote catalog mirror. +type GRPCCatalogWatcherOption func(*GRPCCatalogWatcher) + +// WithGRPCCatalogWatcherRetryInterval sets reconnect and legacy polling delay. +func WithGRPCCatalogWatcherRetryInterval(interval time.Duration) GRPCCatalogWatcherOption { + return func(w *GRPCCatalogWatcher) { + if interval > 0 { + w.retryInterval = interval + } + } +} + +// WithGRPCCatalogWatcherLogger sets the reconnect logger. +func WithGRPCCatalogWatcherLogger(logger *slog.Logger) GRPCCatalogWatcherOption { + return func(w *GRPCCatalogWatcher) { + if logger != nil { + w.logger = logger + } + } +} + +// GRPCCatalogWatcher mirrors a remote Distribution catalog into Engine. +type GRPCCatalogWatcher struct { + clientProvider CatalogWatchClientProvider + engine *Engine + retryInterval time.Duration + logger *slog.Logger +} + +// NewGRPCCatalogWatcher creates a capability-negotiated remote mirror. +func NewGRPCCatalogWatcher(client pb.DistributionClient, engine *Engine, opts ...GRPCCatalogWatcherOption) *GRPCCatalogWatcher { + return NewResolvingGRPCCatalogWatcher(func(context.Context) (pb.DistributionClient, error) { + if client == nil { + return nil, errors.WithStack(ErrCatalogWatchClientRequired) + } + return client, nil + }, engine, opts...) +} + +// NewResolvingGRPCCatalogWatcher creates a mirror that re-resolves its remote +// endpoint before every reconnect attempt. +func NewResolvingGRPCCatalogWatcher( + provider CatalogWatchClientProvider, + engine *Engine, + opts ...GRPCCatalogWatcherOption, +) *GRPCCatalogWatcher { + w := &GRPCCatalogWatcher{ + clientProvider: provider, + engine: engine, + retryInterval: defaultGRPCCatalogWatcherRetryInterval, + logger: slog.Default(), + } + for _, opt := range opts { + if opt != nil { + opt(w) + } + } + return w +} + +// Run reconnects from the last atomically published Engine version. Servers +// without capability negotiation use ListRoutes snapshot polling. +func (w *GRPCCatalogWatcher) Run(ctx context.Context) error { + if err := w.validate(); err != nil { + return err + } + if ctx == nil { + return errors.WithStack(errCatalogWatcherContextRequired) + } + for ctx.Err() == nil { + if err := w.runAttempt(ctx); err != nil && ctx.Err() == nil { + w.logger.WarnContext(ctx, "catalog watch attempt failed", "error", err, "version", w.engine.Version()) + } + if ctx.Err() == nil { + _ = waitCatalogWatcherRetry(ctx, w.retryInterval) + } + } + return nil +} + +func (w *GRPCCatalogWatcher) runAttempt(ctx context.Context) error { + client, err := w.clientProvider(ctx) + if err != nil { + return errors.Wrap(err, "resolve catalog watch endpoint") + } + legacy, maxBatch, err := w.negotiate(ctx, client) + if err != nil { + return errors.Wrap(err, "negotiate catalog watch capability") + } + if legacy { + return errors.Wrap(w.syncSnapshot(ctx, client), "poll legacy catalog snapshot") + } + err = w.consumeStream(ctx, client, maxBatch) + if status.Code(errors.Cause(err)) != codes.Unimplemented { + return err + } + if syncErr := w.syncSnapshot(ctx, client); syncErr != nil { + return errors.Join(err, errors.Wrap(syncErr, "poll catalog snapshot after unimplemented watch")) + } + return nil +} + +func (w *GRPCCatalogWatcher) validate() error { + if w == nil || w.clientProvider == nil { + return errors.WithStack(ErrCatalogWatchClientRequired) + } + if w.engine == nil { + return errors.WithStack(ErrEngineRequired) + } + if w.retryInterval <= 0 { + return errors.WithStack(errCatalogWatcherInvalidInterval) + } + if w.logger == nil { + return errors.WithStack(errCatalogWatcherLoggerRequired) + } + return nil +} + +func (w *GRPCCatalogWatcher) negotiate(ctx context.Context, client pb.DistributionClient) (bool, uint32, error) { + capabilities, err := client.GetCatalogCapabilities( + ctx, + &pb.CatalogCapabilitiesRequest{}, + grpc.WaitForReady(false), + ) + if err != nil { + if status.Code(errors.Cause(err)) == codes.Unimplemented { + return true, 0, nil + } + return false, 0, errors.WithStack(err) + } + if !supportsCatalogWatchProtocol(capabilities.GetSupportedProtocolVersions(), CatalogWatchProtocolVersion) { + return true, 0, nil + } + maxBatch := capabilities.GetMaxBatchSize() + if maxBatch == 0 || maxBatch > MaxCatalogDeltaBatchSize { + maxBatch = MaxCatalogDeltaBatchSize + } + return false, maxBatch, nil +} + +func supportsCatalogWatchProtocol(supported []uint32, target uint32) bool { + for _, version := range supported { + if version == target { + return true + } + } + return false +} + +func (w *GRPCCatalogWatcher) consumeStream(ctx context.Context, client pb.DistributionClient, maxBatch uint32) error { + streamCtx, cancel := context.WithCancel(ctx) + defer cancel() + stream, err := client.WatchCatalog(streamCtx, &pb.CatalogWatchRequest{ + ProtocolVersion: CatalogWatchProtocolVersion, + AfterVersion: w.engine.Version(), + MaxBatchSize: maxBatch, + }, grpc.WaitForReady(false)) + if err != nil { + return errors.WithStack(err) + } + for { + event, err := stream.Recv() + if err != nil { + if errors.Is(err, io.EOF) { + return io.EOF + } + return errors.WithStack(err) + } + if err := w.applyEvent(event); err != nil { + if errors.Is(err, ErrEngineDeltaVersionGap) || errors.Is(err, ErrCatalogDeltaVersionGap) { + return w.syncSnapshot(ctx, client) + } + return err + } + } +} + +func (w *GRPCCatalogWatcher) applyEvent(event *pb.CatalogWatchEvent) error { + if event == nil { + return errors.WithStack(ErrCatalogWatchEventInvalid) + } + if snapshot := event.GetSnapshot(); snapshot != nil { + routes, err := routeDescriptorsFromProto(snapshot.GetRoutes()) + if err != nil { + return err + } + if err := w.engine.ApplySnapshot(CatalogSnapshot{Version: snapshot.GetVersion(), Routes: routes}); err != nil { + if errors.Is(err, ErrEngineSnapshotVersionStale) { + return nil + } + return err + } + return nil + } + if record := event.GetDelta(); record != nil { + delta, err := catalogDeltaFromProto(record) + if err != nil { + return err + } + if err := w.engine.ApplyDelta(delta); err != nil { + if errors.Is(err, ErrEngineSnapshotVersionStale) { + return nil + } + return err + } + return nil + } + return errors.WithStack(ErrCatalogWatchEventInvalid) +} + +func (w *GRPCCatalogWatcher) syncSnapshot(ctx context.Context, client pb.DistributionClient) error { + response, err := client.ListRoutes(ctx, &pb.ListRoutesRequest{}, grpc.WaitForReady(false)) + if err != nil { + return errors.WithStack(err) + } + routes, err := routeDescriptorsFromProto(response.GetRoutes()) + if err != nil { + return err + } + if err := w.engine.ApplySnapshot(CatalogSnapshot{Version: response.GetCatalogVersion(), Routes: routes}); err != nil { + if errors.Is(err, ErrEngineSnapshotVersionStale) { + return nil + } + return err + } + return nil +} + +func waitCatalogWatcherRetry(ctx context.Context, interval time.Duration) error { + timer := time.NewTimer(interval) + defer timer.Stop() + select { + case <-ctx.Done(): + return errors.WithStack(ctx.Err()) + case <-timer.C: + return nil + } +} + +func catalogDeltaFromProto(record *pb.CatalogDeltaRecord) (CatalogDelta, error) { + delta := CatalogDelta{ + PreviousVersion: record.GetPreviousVersion(), + Version: record.GetVersion(), + Mutations: make([]CatalogRouteMutation, 0, len(record.GetMutations())), + } + for _, raw := range record.GetMutations() { + if raw == nil { + return CatalogDelta{}, errors.WithStack(ErrCatalogWatchEventInvalid) + } + mutation := CatalogRouteMutation{RouteID: raw.GetRouteId()} + switch raw.GetOp() { + case pb.CatalogDeltaMutationOp_CATALOG_DELTA_MUTATION_OP_UNSPECIFIED: + return CatalogDelta{}, errors.WithStack(ErrCatalogWatchEventInvalid) + case pb.CatalogDeltaMutationOp_CATALOG_DELTA_MUTATION_OP_DELETE: + if raw.Route != nil { + return CatalogDelta{}, errors.WithStack(ErrCatalogWatchEventInvalid) + } + mutation.Op = CatalogMutationDelete + case pb.CatalogDeltaMutationOp_CATALOG_DELTA_MUTATION_OP_UPSERT: + mutation.Op = CatalogMutationUpsert + route, err := routeDescriptorFromProto(raw.GetRoute()) + if err != nil { + return CatalogDelta{}, err + } + mutation.Route = route + default: + return CatalogDelta{}, errors.WithStack(ErrCatalogWatchEventInvalid) + } + delta.Mutations = append(delta.Mutations, mutation) + } + if err := validateCatalogDelta(delta); err != nil { + return CatalogDelta{}, err + } + return delta, nil +} + +func routeDescriptorsFromProto(routes []*pb.RouteDescriptor) ([]RouteDescriptor, error) { + out := make([]RouteDescriptor, 0, len(routes)) + for _, raw := range routes { + route, err := routeDescriptorFromProto(raw) + if err != nil { + return nil, err + } + out = append(out, route) + } + return out, nil +} + +func routeDescriptorFromProto(raw *pb.RouteDescriptor) (RouteDescriptor, error) { + if raw == nil { + return RouteDescriptor{}, errors.WithStack(ErrCatalogWatchEventInvalid) + } + state, ok := routeStateFromProto(raw.GetState()) + if !ok { + return RouteDescriptor{}, errors.WithStack(ErrCatalogWatchEventInvalid) + } + route := RouteDescriptor{ + RouteID: raw.GetRouteId(), + Start: CloneBytes(raw.GetStart()), + End: CloneBytes(raw.GetEnd()), + GroupID: raw.GetRaftGroupId(), + State: state, + ParentRouteID: raw.GetParentRouteId(), + } + if err := validateRouteDescriptor(route); err != nil { + return RouteDescriptor{}, err + } + return route, nil +} + +func routeStateFromProto(state pb.RouteState) (RouteState, bool) { + switch state { + case pb.RouteState_ROUTE_STATE_UNSPECIFIED: + return 0, false + case pb.RouteState_ROUTE_STATE_ACTIVE: + return RouteStateActive, true + case pb.RouteState_ROUTE_STATE_WRITE_FENCED: + return RouteStateWriteFenced, true + case pb.RouteState_ROUTE_STATE_MIGRATING_SOURCE: + return RouteStateMigratingSource, true + case pb.RouteState_ROUTE_STATE_MIGRATING_TARGET: + return RouteStateMigratingTarget, true + default: + return 0, false + } +} diff --git a/distribution/watcher.go b/distribution/watcher.go index decd21afb..bb4ae095f 100644 --- a/distribution/watcher.go +++ b/distribution/watcher.go @@ -37,22 +37,34 @@ func WithCatalogWatcherLogger(logger *slog.Logger) CatalogWatcherOption { } } +// WithCatalogWatcherBatchSize sets the maximum number of deltas applied by one +// synchronization pass. +func WithCatalogWatcherBatchSize(batchSize int) CatalogWatcherOption { + return func(w *CatalogWatcher) { + if batchSize > 0 { + w.batchSize = batchSize + } + } +} + // CatalogWatcher periodically refreshes Engine from durable catalog snapshots. type CatalogWatcher struct { - catalog *CatalogStore - engine *Engine - interval time.Duration - logger *slog.Logger + catalog *CatalogStore + engine *Engine + interval time.Duration + batchSize int + logger *slog.Logger } // NewCatalogWatcher creates a watcher that polls the durable route catalog and // applies newer snapshots to the in-memory engine. func NewCatalogWatcher(catalog *CatalogStore, engine *Engine, opts ...CatalogWatcherOption) *CatalogWatcher { w := &CatalogWatcher{ - catalog: catalog, - engine: engine, - interval: defaultCatalogWatcherInterval, - logger: slog.Default(), + catalog: catalog, + engine: engine, + interval: defaultCatalogWatcherInterval, + batchSize: DefaultCatalogDeltaBatchSize, + logger: slog.Default(), } for _, opt := range opts { if opt != nil { @@ -110,29 +122,26 @@ func (w *CatalogWatcher) SyncOnce(ctx context.Context) error { return errors.WithStack(errCatalogWatcherContextRequired) } - readTS := w.catalog.store.LastCommitTS() - catalogVersion, err := w.catalog.versionAt(ctx, readTS) + changes, err := w.catalog.ChangesSince(ctx, w.engine.Version(), w.batchSize) if err != nil { return err } - if catalogVersion <= w.engine.Version() { + if changes.Reset != nil { + if err := w.engine.ApplySnapshot(*changes.Reset); err != nil { + if errors.Is(err, ErrEngineSnapshotVersionStale) { + return nil + } + return err + } return nil } - - routes, err := w.catalog.routesAt(ctx, readTS) - if err != nil { - return err - } - snapshot := CatalogSnapshot{ - Version: catalogVersion, - Routes: routes, - ReadTS: readTS, - } - if err := w.engine.ApplySnapshot(snapshot); err != nil { - if errors.Is(err, ErrEngineSnapshotVersionStale) { - return nil + for _, delta := range changes.Deltas { + if err := w.engine.ApplyDelta(delta); err != nil { + if errors.Is(err, ErrEngineSnapshotVersionStale) { + continue + } + return err } - return err } return nil } @@ -147,6 +156,9 @@ func (w *CatalogWatcher) validate() error { if w.interval <= 0 { return errors.WithStack(errCatalogWatcherInvalidInterval) } + if w.batchSize <= 0 { + return errors.WithStack(ErrCatalogDeltaLimitInvalid) + } if w.logger == nil { return errors.WithStack(errCatalogWatcherLoggerRequired) } diff --git a/docs/design/2026_07_18_implemented_route_catalog_delta_watch.md b/docs/design/2026_07_18_implemented_route_catalog_delta_watch.md new file mode 100644 index 000000000..3c7659eaa --- /dev/null +++ b/docs/design/2026_07_18_implemented_route_catalog_delta_watch.md @@ -0,0 +1,161 @@ +# Versioned route catalog delta and streaming watch + +Status: Implemented +Author: bootjp +Date: 2026-07-18 + +## 1. Scope + +This design replaces full-catalog refresh as the steady-state propagation path +for route catalog changes. It owns: + +- a durable, versioned route delta log; +- bounded retention with full-snapshot reset for stale cursors; +- atomic publication into each node's in-memory route engine; +- capability negotiation and a server-streaming gRPC watch; +- reconnect from the last published version; +- rolling-upgrade fallback to the existing full-snapshot poll. + +The route index and batched catalog mutation designs remain separate. This +change preserves the existing one-version-per-catalog-commit contract and the +route engine's slice representation. + +## 2. Durable representation + +Each catalog transition from version `v` to `v+1` writes all of the following +in the same MVCC transaction: + +- changed route rows; +- deleted route rows; +- `!dist|delta|` containing the ordered route + mutations; +- `!dist|meta|delta_floor`, the oldest retained delta version; +- `!dist|meta|version`, the new catalog version. + +A delta records exactly one contiguous transition: + +```text +CatalogDelta { + previous_version: v + version: v + 1 + mutations: UPSERT(RouteDescriptor) | DELETE(route_id) +} +``` + +Mutation route IDs are unique within a delta. Upserts carry a validated route +whose ID matches the mutation ID. Deletes carry no route payload. The codec is +versioned and rejects trailing bytes, oversized lengths, unknown operations, +zero route IDs, and non-contiguous versions. + +`CatalogStore.Save` derives the delta from the old and desired route sets. +`DistributionServer.SplitRange` constructs the equivalent parent delete and +two child upserts and includes them in the coordinator transaction that commits +the route rows and catalog version. A successful catalog version therefore +cannot be visible without its matching delta. + +## 3. Retention and reconnect + +The default retention window is 1,024 catalog transitions. Delta version keys +sort by version and the floor advances as old entries are deleted in the same +catalog commit. + +`CatalogStore.ChangesSince(after_version, limit)` reads the version, floor, +and deltas at one MVCC timestamp. It returns one of: + +- no change when `after_version` equals the durable version; +- a bounded contiguous delta batch; +- a full `CatalogSnapshot` reset when the requested next version predates the + retained floor or the catalog predates delta logging. + +A cursor ahead of the durable version is rejected. A missing or malformed +delta inside the advertised retained range is an integrity error, not a silent +snapshot fallback. + +## 4. Atomic mirror publication + +`Engine.ApplyDelta` builds the complete next route table while holding the +engine write lock, validates route order and overlap, and only then replaces the +published slice and version. Readers observe either the complete previous +version or the complete next version. A gap, malformed mutation, or invalid +route leaves both the route table and version unchanged. + +The first durable delta removes the process bootstrap route with ID zero. +Upserts preserve the current in-memory load counter for an existing route. +Successful publication records the same immutable history snapshot used by the +cross-version read gate. + +The local `CatalogWatcher` consumes bounded delta batches. Retention reset uses +the existing atomic `ApplySnapshot` path. Stale work is harmless when the local +poller and remote stream race because both publication paths reject version +regression. + +## 5. Streaming protocol + +`proto.Distribution` exposes: + +- `GetCatalogCapabilities`, which advertises supported watch protocol versions, + the current version, oldest retained delta, and maximum batch size; +- `WatchCatalog`, which accepts a protocol version, reconnect cursor, and batch + limit and streams either `CatalogDeltaRecord` or `CatalogSnapshotReset`. + +Protocol version 1 requires contiguous one-version deltas. The server caps a +client batch request at 1,024 and uses 128 when the client omits a limit. +Unsupported protocols fail with `FailedPrecondition`; future cursors fail with +`InvalidArgument`. + +Production resolves the current catalog-group leader before each stream +attempt. The leader closes the stream with `Unavailable` after it loses +leadership, causing the client to resolve the new leader and reconnect from the +last version atomically published in its engine. + +The existing 100 ms local durable watcher remains active while the stream is +connected and while it is unavailable. It bounds staleness during elections, +network failure, and mixed-version rollout. A server without capability or +watch support causes the client to use `ListRoutes` polling and negotiate again +on the next attempt, so it can adopt streaming after the rolling upgrade +finishes. + +## 6. Failure and rollout behavior + +- Stream disconnect loses no catalog transition because reconnect uses the + mirror's published version and deltas are durable. +- Falling behind retention produces an explicit snapshot reset. +- Corruption or a hole within retention fails closed and is surfaced for + operator investigation. +- A follower does not retain leader-targeted watch streams. +- Old clients continue using `ListRoutes`; old servers return `Unimplemented` + for the new RPCs and new clients retain full-snapshot polling. +- The stream is an optimization. Route correctness continues to depend on the + durable catalog-group state and monotonic engine publication. + +## 7. Verification + +Focused package tests: + +```sh +go test ./distribution -run 'Test(CatalogDelta|CatalogStore|EngineApplyDelta|CatalogWatcherAppliesBounded)' -count=1 +go test ./adapter -run 'Test(DistributionServerCatalogWatch|GRPCCatalogWatcher|CatalogStoreMutationsToOps)' -count=1 +``` + +The focused coverage includes: + +- codec round trip and invalid record rejection; +- contiguous `CatalogStore.Save` deltas; +- retention deletion and snapshot reset; +- missing retained-delta integrity failure; +- engine gap and overlap rejection without partial publication; +- bounded local watcher batches; +- split transaction delta visibility; +- protocol negotiation, reconnect cursor, and snapshot stream events; +- disconnect followed by endpoint re-resolution; +- follower stream rejection; +- old-server full-snapshot fallback. + +Repository gates: + +```sh +go test ./... +go test -race ./distribution ./adapter +GOCACHE=$(pwd)/.cache GOLANGCI_LINT_CACHE=$(pwd)/.golangci-cache golangci-lint run ./... --timeout=5m +git diff --check +``` diff --git a/main.go b/main.go index 479bcc69b..671e69fc4 100644 --- a/main.go +++ b/main.go @@ -513,10 +513,6 @@ func run() error { WithKeyVizLabelsEnabled(*keyvizLabelsEnabled). WithAllShardGroups(dataGroupIDs(cfg.groups)...). WithPartitionResolver(buildSQSPartitionResolver(cfg.sqsFifoPartitionMap)) - if err := configureCoordinatorTSO(coordinate); err != nil { - return err - } - // SQS HT-FIFO ยง8 leadership-refusal: install per-group // observers that step the local node down via // TransferLeadership when it acquires (or already holds) @@ -537,9 +533,10 @@ func run() error { // registry-read / behind-epoch failure fails the process // synchronously here, BEFORE the gRPC servers serve, so writes never // run with no registration gate installed. - distCatalog, err := setupDistributionAndRegistration( + distCatalog, catalogRuntime, defaultRuntime, err := setupDistributionRuntimeDependencies( runCtx, eg, runtimes, cfg.engine, - coordinate, shardGroups[cfg.defaultGroup], encWiring, *raftId, *encryptionSidecarPath) + coordinate, shardGroups[cfg.defaultGroup], cfg.defaultGroup, + encWiring, *raftId, *encryptionSidecarPath) if err != nil { cancel() return err @@ -552,9 +549,16 @@ func run() error { // every dispatched mutation. seedKeyVizRoutes(sampler, cfg.engine) + // The local durable watcher is the 100 ms fallback required when the + // best-effort leader stream is unavailable or reconnecting. eg.Go(func() error { return runDistributionCatalogWatcher(runCtx, distCatalog, cfg.engine) }) + catalogWatchConns := &kv.GRPCConnCache{} + cleanup.Add(func() { _ = catalogWatchConns.Close() }) + eg.Go(func() error { + return runDistributionCatalogStream(runCtx, catalogRuntime, catalogWatchConns, cfg.engine) + }) startKeyVizFlusher(runCtx, eg, sampler) startKeyVizLeaderTermPublisher(runCtx, eg, sampler, runtimes) startMemoryWatchdog(runCtx, eg, cancel) @@ -563,6 +567,10 @@ func run() error { distCatalog, adapter.WithDistributionCoordinator(coordinate), adapter.WithDistributionActiveTimestampTracker(readTracker), + adapter.WithCatalogWatchLeaderCheck(func() bool { + engine := catalogRuntime.snapshotEngine() + return engine != nil && engine.State() == raftengine.StateLeader + }), adapter.WithDistributionFilesystemObserver(metricsRegistry.FileSystemObserver()), ) startMonitoringCollectors(runCtx, metricsRegistry, runtimes, clock) @@ -575,7 +583,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, @@ -2855,6 +2862,77 @@ func runDistributionCatalogWatcher(ctx context.Context, catalog *distribution.Ca return nil } +func setupDistributionRuntimeDependencies( + runCtx context.Context, + eg *errgroup.Group, + runtimes []*raftGroupRuntime, + engine *distribution.Engine, + coordinate *kv.ShardedCoordinator, + defaultGroup *kv.ShardGroup, + defaultGroupID uint64, + encWiring encryptionWriteWiring, + raftID string, + sidecarPath string, +) (*distribution.CatalogStore, *raftGroupRuntime, *raftGroupRuntime, error) { + if err := configureCoordinatorTSO(coordinate); err != nil { + return nil, nil, nil, err + } + catalog, err := setupDistributionAndRegistration( + runCtx, eg, runtimes, engine, coordinate, defaultGroup, encWiring, raftID, sidecarPath) + if err != nil { + return nil, nil, nil, err + } + catalogGroupID, err := distributionCatalogGroupID(engine) + if err != nil { + return nil, nil, nil, errors.Wrap(err, "resolve distribution catalog group runtime") + } + catalogRuntime := findDefaultGroupRuntime(runtimes, catalogGroupID) + if catalogRuntime == nil { + return nil, nil, nil, errors.WithStack(errors.Newf( + "distribution catalog raft group %d runtime is unavailable", + catalogGroupID, + )) + } + defaultRuntime := findDefaultGroupRuntime(runtimes, defaultGroupID) + if defaultRuntime == nil { + return nil, nil, nil, errors.WithStack(errors.Newf( + "default raft group %d runtime is unavailable", + defaultGroupID, + )) + } + return catalog, catalogRuntime, defaultRuntime, nil +} + +func runDistributionCatalogStream( + ctx context.Context, + runtime *raftGroupRuntime, + conns *kv.GRPCConnCache, + engine *distribution.Engine, +) error { + watcher := distribution.NewResolvingGRPCCatalogWatcher( + func(context.Context) (pb.DistributionClient, error) { + raftEngine := runtime.snapshotEngine() + if raftEngine == nil { + return nil, errors.New("default raft group engine is unavailable") + } + leader := raftEngine.Leader().Address + if leader == "" { + return nil, errors.New("default raft group leader is unavailable") + } + conn, err := conns.ConnFor(leader) + if err != nil { + return nil, errors.WithStack(err) + } + return pb.NewDistributionClient(conn), nil + }, + engine, + ) + if err := watcher.Run(ctx); err != nil { + return errors.Wrap(err, "catalog stream watcher failed") + } + return nil +} + func waitErrgroupAfterStartupFailure(cancel context.CancelFunc, eg *errgroup.Group, startupErr error) error { cancel() if err := eg.Wait(); err != nil { diff --git a/proto/distribution.pb.go b/proto/distribution.pb.go index b6847386e..14a3db51f 100644 --- a/proto/distribution.pb.go +++ b/proto/distribution.pb.go @@ -250,6 +250,55 @@ func (SplitJobExportPhase) EnumDescriptor() ([]byte, []int) { return file_distribution_proto_rawDescGZIP(), []int{3} } +type CatalogDeltaMutationOp int32 + +const ( + CatalogDeltaMutationOp_CATALOG_DELTA_MUTATION_OP_UNSPECIFIED CatalogDeltaMutationOp = 0 + CatalogDeltaMutationOp_CATALOG_DELTA_MUTATION_OP_UPSERT CatalogDeltaMutationOp = 1 + CatalogDeltaMutationOp_CATALOG_DELTA_MUTATION_OP_DELETE CatalogDeltaMutationOp = 2 +) + +// Enum value maps for CatalogDeltaMutationOp. +var ( + CatalogDeltaMutationOp_name = map[int32]string{ + 0: "CATALOG_DELTA_MUTATION_OP_UNSPECIFIED", + 1: "CATALOG_DELTA_MUTATION_OP_UPSERT", + 2: "CATALOG_DELTA_MUTATION_OP_DELETE", + } + CatalogDeltaMutationOp_value = map[string]int32{ + "CATALOG_DELTA_MUTATION_OP_UNSPECIFIED": 0, + "CATALOG_DELTA_MUTATION_OP_UPSERT": 1, + "CATALOG_DELTA_MUTATION_OP_DELETE": 2, + } +) + +func (x CatalogDeltaMutationOp) Enum() *CatalogDeltaMutationOp { + p := new(CatalogDeltaMutationOp) + *p = x + return p +} + +func (x CatalogDeltaMutationOp) String() string { + return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x)) +} + +func (CatalogDeltaMutationOp) Descriptor() protoreflect.EnumDescriptor { + return file_distribution_proto_enumTypes[4].Descriptor() +} + +func (CatalogDeltaMutationOp) Type() protoreflect.EnumType { + return &file_distribution_proto_enumTypes[4] +} + +func (x CatalogDeltaMutationOp) Number() protoreflect.EnumNumber { + return protoreflect.EnumNumber(x) +} + +// Deprecated: Use CatalogDeltaMutationOp.Descriptor instead. +func (CatalogDeltaMutationOp) EnumDescriptor() ([]byte, []int) { + return file_distribution_proto_rawDescGZIP(), []int{4} +} + type GetRouteRequest struct { state protoimpl.MessageState `protogen:"open.v1"` Key []byte `protobuf:"bytes,1,opt,name=key,proto3" json:"key,omitempty"` @@ -1136,6 +1185,424 @@ func (x *SplitRangeResponse) GetRight() *RouteDescriptor { return nil } +type CatalogCapabilitiesRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *CatalogCapabilitiesRequest) Reset() { + *x = CatalogCapabilitiesRequest{} + mi := &file_distribution_proto_msgTypes[11] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *CatalogCapabilitiesRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*CatalogCapabilitiesRequest) ProtoMessage() {} + +func (x *CatalogCapabilitiesRequest) ProtoReflect() protoreflect.Message { + mi := &file_distribution_proto_msgTypes[11] + 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 CatalogCapabilitiesRequest.ProtoReflect.Descriptor instead. +func (*CatalogCapabilitiesRequest) Descriptor() ([]byte, []int) { + return file_distribution_proto_rawDescGZIP(), []int{11} +} + +type CatalogCapabilitiesResponse struct { + state protoimpl.MessageState `protogen:"open.v1"` + SupportedProtocolVersions []uint32 `protobuf:"varint,1,rep,packed,name=supported_protocol_versions,json=supportedProtocolVersions,proto3" json:"supported_protocol_versions,omitempty"` + CurrentVersion uint64 `protobuf:"varint,2,opt,name=current_version,json=currentVersion,proto3" json:"current_version,omitempty"` + OldestDeltaVersion uint64 `protobuf:"varint,3,opt,name=oldest_delta_version,json=oldestDeltaVersion,proto3" json:"oldest_delta_version,omitempty"` + MaxBatchSize uint32 `protobuf:"varint,4,opt,name=max_batch_size,json=maxBatchSize,proto3" json:"max_batch_size,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *CatalogCapabilitiesResponse) Reset() { + *x = CatalogCapabilitiesResponse{} + mi := &file_distribution_proto_msgTypes[12] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *CatalogCapabilitiesResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*CatalogCapabilitiesResponse) ProtoMessage() {} + +func (x *CatalogCapabilitiesResponse) ProtoReflect() protoreflect.Message { + mi := &file_distribution_proto_msgTypes[12] + 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 CatalogCapabilitiesResponse.ProtoReflect.Descriptor instead. +func (*CatalogCapabilitiesResponse) Descriptor() ([]byte, []int) { + return file_distribution_proto_rawDescGZIP(), []int{12} +} + +func (x *CatalogCapabilitiesResponse) GetSupportedProtocolVersions() []uint32 { + if x != nil { + return x.SupportedProtocolVersions + } + return nil +} + +func (x *CatalogCapabilitiesResponse) GetCurrentVersion() uint64 { + if x != nil { + return x.CurrentVersion + } + return 0 +} + +func (x *CatalogCapabilitiesResponse) GetOldestDeltaVersion() uint64 { + if x != nil { + return x.OldestDeltaVersion + } + return 0 +} + +func (x *CatalogCapabilitiesResponse) GetMaxBatchSize() uint32 { + if x != nil { + return x.MaxBatchSize + } + return 0 +} + +type CatalogWatchRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + ProtocolVersion uint32 `protobuf:"varint,1,opt,name=protocol_version,json=protocolVersion,proto3" json:"protocol_version,omitempty"` + AfterVersion uint64 `protobuf:"varint,2,opt,name=after_version,json=afterVersion,proto3" json:"after_version,omitempty"` + MaxBatchSize uint32 `protobuf:"varint,3,opt,name=max_batch_size,json=maxBatchSize,proto3" json:"max_batch_size,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *CatalogWatchRequest) Reset() { + *x = CatalogWatchRequest{} + mi := &file_distribution_proto_msgTypes[13] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *CatalogWatchRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*CatalogWatchRequest) ProtoMessage() {} + +func (x *CatalogWatchRequest) 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 CatalogWatchRequest.ProtoReflect.Descriptor instead. +func (*CatalogWatchRequest) Descriptor() ([]byte, []int) { + return file_distribution_proto_rawDescGZIP(), []int{13} +} + +func (x *CatalogWatchRequest) GetProtocolVersion() uint32 { + if x != nil { + return x.ProtocolVersion + } + return 0 +} + +func (x *CatalogWatchRequest) GetAfterVersion() uint64 { + if x != nil { + return x.AfterVersion + } + return 0 +} + +func (x *CatalogWatchRequest) GetMaxBatchSize() uint32 { + if x != nil { + return x.MaxBatchSize + } + return 0 +} + +type CatalogDeltaMutation struct { + state protoimpl.MessageState `protogen:"open.v1"` + Op CatalogDeltaMutationOp `protobuf:"varint,1,opt,name=op,proto3,enum=CatalogDeltaMutationOp" json:"op,omitempty"` + RouteId uint64 `protobuf:"varint,2,opt,name=route_id,json=routeId,proto3" json:"route_id,omitempty"` + Route *RouteDescriptor `protobuf:"bytes,3,opt,name=route,proto3" json:"route,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *CatalogDeltaMutation) Reset() { + *x = CatalogDeltaMutation{} + mi := &file_distribution_proto_msgTypes[14] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *CatalogDeltaMutation) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*CatalogDeltaMutation) ProtoMessage() {} + +func (x *CatalogDeltaMutation) 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 CatalogDeltaMutation.ProtoReflect.Descriptor instead. +func (*CatalogDeltaMutation) Descriptor() ([]byte, []int) { + return file_distribution_proto_rawDescGZIP(), []int{14} +} + +func (x *CatalogDeltaMutation) GetOp() CatalogDeltaMutationOp { + if x != nil { + return x.Op + } + return CatalogDeltaMutationOp_CATALOG_DELTA_MUTATION_OP_UNSPECIFIED +} + +func (x *CatalogDeltaMutation) GetRouteId() uint64 { + if x != nil { + return x.RouteId + } + return 0 +} + +func (x *CatalogDeltaMutation) GetRoute() *RouteDescriptor { + if x != nil { + return x.Route + } + return nil +} + +type CatalogDeltaRecord struct { + state protoimpl.MessageState `protogen:"open.v1"` + PreviousVersion uint64 `protobuf:"varint,1,opt,name=previous_version,json=previousVersion,proto3" json:"previous_version,omitempty"` + Version uint64 `protobuf:"varint,2,opt,name=version,proto3" json:"version,omitempty"` + Mutations []*CatalogDeltaMutation `protobuf:"bytes,3,rep,name=mutations,proto3" json:"mutations,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *CatalogDeltaRecord) Reset() { + *x = CatalogDeltaRecord{} + mi := &file_distribution_proto_msgTypes[15] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *CatalogDeltaRecord) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*CatalogDeltaRecord) ProtoMessage() {} + +func (x *CatalogDeltaRecord) ProtoReflect() protoreflect.Message { + mi := &file_distribution_proto_msgTypes[15] + 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 CatalogDeltaRecord.ProtoReflect.Descriptor instead. +func (*CatalogDeltaRecord) Descriptor() ([]byte, []int) { + return file_distribution_proto_rawDescGZIP(), []int{15} +} + +func (x *CatalogDeltaRecord) GetPreviousVersion() uint64 { + if x != nil { + return x.PreviousVersion + } + return 0 +} + +func (x *CatalogDeltaRecord) GetVersion() uint64 { + if x != nil { + return x.Version + } + return 0 +} + +func (x *CatalogDeltaRecord) GetMutations() []*CatalogDeltaMutation { + if x != nil { + return x.Mutations + } + return nil +} + +type CatalogSnapshotReset struct { + state protoimpl.MessageState `protogen:"open.v1"` + Version uint64 `protobuf:"varint,1,opt,name=version,proto3" json:"version,omitempty"` + Routes []*RouteDescriptor `protobuf:"bytes,2,rep,name=routes,proto3" json:"routes,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *CatalogSnapshotReset) Reset() { + *x = CatalogSnapshotReset{} + mi := &file_distribution_proto_msgTypes[16] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *CatalogSnapshotReset) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*CatalogSnapshotReset) ProtoMessage() {} + +func (x *CatalogSnapshotReset) ProtoReflect() protoreflect.Message { + mi := &file_distribution_proto_msgTypes[16] + 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 CatalogSnapshotReset.ProtoReflect.Descriptor instead. +func (*CatalogSnapshotReset) Descriptor() ([]byte, []int) { + return file_distribution_proto_rawDescGZIP(), []int{16} +} + +func (x *CatalogSnapshotReset) GetVersion() uint64 { + if x != nil { + return x.Version + } + return 0 +} + +func (x *CatalogSnapshotReset) GetRoutes() []*RouteDescriptor { + if x != nil { + return x.Routes + } + return nil +} + +type CatalogWatchEvent struct { + state protoimpl.MessageState `protogen:"open.v1"` + // Types that are valid to be assigned to Payload: + // + // *CatalogWatchEvent_Snapshot + // *CatalogWatchEvent_Delta + Payload isCatalogWatchEvent_Payload `protobuf_oneof:"payload"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *CatalogWatchEvent) Reset() { + *x = CatalogWatchEvent{} + mi := &file_distribution_proto_msgTypes[17] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *CatalogWatchEvent) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*CatalogWatchEvent) ProtoMessage() {} + +func (x *CatalogWatchEvent) ProtoReflect() protoreflect.Message { + mi := &file_distribution_proto_msgTypes[17] + 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 CatalogWatchEvent.ProtoReflect.Descriptor instead. +func (*CatalogWatchEvent) Descriptor() ([]byte, []int) { + return file_distribution_proto_rawDescGZIP(), []int{17} +} + +func (x *CatalogWatchEvent) GetPayload() isCatalogWatchEvent_Payload { + if x != nil { + return x.Payload + } + return nil +} + +func (x *CatalogWatchEvent) GetSnapshot() *CatalogSnapshotReset { + if x != nil { + if x, ok := x.Payload.(*CatalogWatchEvent_Snapshot); ok { + return x.Snapshot + } + } + return nil +} + +func (x *CatalogWatchEvent) GetDelta() *CatalogDeltaRecord { + if x != nil { + if x, ok := x.Payload.(*CatalogWatchEvent_Delta); ok { + return x.Delta + } + } + return nil +} + +type isCatalogWatchEvent_Payload interface { + isCatalogWatchEvent_Payload() +} + +type CatalogWatchEvent_Snapshot struct { + Snapshot *CatalogSnapshotReset `protobuf:"bytes,1,opt,name=snapshot,proto3,oneof"` +} + +type CatalogWatchEvent_Delta struct { + Delta *CatalogDeltaRecord `protobuf:"bytes,2,opt,name=delta,proto3,oneof"` +} + +func (*CatalogWatchEvent_Snapshot) isCatalogWatchEvent_Payload() {} + +func (*CatalogWatchEvent_Delta) isCatalogWatchEvent_Payload() {} + var File_distribution_proto protoreflect.FileDescriptor const file_distribution_proto_rawDesc = "" + @@ -1219,7 +1686,32 @@ const file_distribution_proto_rawDesc = "" + "\x12SplitRangeResponse\x12'\n" + "\x0fcatalog_version\x18\x01 \x01(\x04R\x0ecatalogVersion\x12$\n" + "\x04left\x18\x02 \x01(\v2\x10.RouteDescriptorR\x04left\x12&\n" + - "\x05right\x18\x03 \x01(\v2\x10.RouteDescriptorR\x05right*\xa3\x01\n" + + "\x05right\x18\x03 \x01(\v2\x10.RouteDescriptorR\x05right\"\x1c\n" + + "\x1aCatalogCapabilitiesRequest\"\xde\x01\n" + + "\x1bCatalogCapabilitiesResponse\x12>\n" + + "\x1bsupported_protocol_versions\x18\x01 \x03(\rR\x19supportedProtocolVersions\x12'\n" + + "\x0fcurrent_version\x18\x02 \x01(\x04R\x0ecurrentVersion\x120\n" + + "\x14oldest_delta_version\x18\x03 \x01(\x04R\x12oldestDeltaVersion\x12$\n" + + "\x0emax_batch_size\x18\x04 \x01(\rR\fmaxBatchSize\"\x8b\x01\n" + + "\x13CatalogWatchRequest\x12)\n" + + "\x10protocol_version\x18\x01 \x01(\rR\x0fprotocolVersion\x12#\n" + + "\rafter_version\x18\x02 \x01(\x04R\fafterVersion\x12$\n" + + "\x0emax_batch_size\x18\x03 \x01(\rR\fmaxBatchSize\"\x82\x01\n" + + "\x14CatalogDeltaMutation\x12'\n" + + "\x02op\x18\x01 \x01(\x0e2\x17.CatalogDeltaMutationOpR\x02op\x12\x19\n" + + "\broute_id\x18\x02 \x01(\x04R\arouteId\x12&\n" + + "\x05route\x18\x03 \x01(\v2\x10.RouteDescriptorR\x05route\"\x8e\x01\n" + + "\x12CatalogDeltaRecord\x12)\n" + + "\x10previous_version\x18\x01 \x01(\x04R\x0fpreviousVersion\x12\x18\n" + + "\aversion\x18\x02 \x01(\x04R\aversion\x123\n" + + "\tmutations\x18\x03 \x03(\v2\x15.CatalogDeltaMutationR\tmutations\"Z\n" + + "\x14CatalogSnapshotReset\x12\x18\n" + + "\aversion\x18\x01 \x01(\x04R\aversion\x12(\n" + + "\x06routes\x18\x02 \x03(\v2\x10.RouteDescriptorR\x06routes\"\x80\x01\n" + + "\x11CatalogWatchEvent\x123\n" + + "\bsnapshot\x18\x01 \x01(\v2\x15.CatalogSnapshotResetH\x00R\bsnapshot\x12+\n" + + "\x05delta\x18\x02 \x01(\v2\x13.CatalogDeltaRecordH\x00R\x05deltaB\t\n" + + "\apayload*\xa3\x01\n" + "\n" + "RouteState\x12\x1b\n" + "\x17ROUTE_STATE_UNSPECIFIED\x10\x00\x12\x16\n" + @@ -1248,14 +1740,20 @@ 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\xf2\x01\n" + + "!SPLIT_JOB_EXPORT_PHASE_DELTA_COPY\x10\x02*\x8f\x01\n" + + "\x16CatalogDeltaMutationOp\x12)\n" + + "%CATALOG_DELTA_MUTATION_OP_UNSPECIFIED\x10\x00\x12$\n" + + " CATALOG_DELTA_MUTATION_OP_UPSERT\x10\x01\x12$\n" + + " CATALOG_DELTA_MUTATION_OP_DELETE\x10\x022\x87\x03\n" + "\fDistribution\x121\n" + "\bGetRoute\x12\x10.GetRouteRequest\x1a\x11.GetRouteResponse\"\x00\x12=\n" + "\fGetTimestamp\x12\x14.GetTimestampRequest\x1a\x15.GetTimestampResponse\"\x00\x127\n" + "\n" + "ListRoutes\x12\x12.ListRoutesRequest\x1a\x13.ListRoutesResponse\"\x00\x127\n" + "\n" + - "SplitRange\x12\x12.SplitRangeRequest\x1a\x13.SplitRangeResponse\"\x00B#Z!github.com/bootjp/elastickv/protob\x06proto3" + "SplitRange\x12\x12.SplitRangeRequest\x1a\x13.SplitRangeResponse\"\x00\x12U\n" + + "\x16GetCatalogCapabilities\x12\x1b.CatalogCapabilitiesRequest\x1a\x1c.CatalogCapabilitiesResponse\"\x00\x12<\n" + + "\fWatchCatalog\x12\x14.CatalogWatchRequest\x1a\x12.CatalogWatchEvent\"\x000\x01B#Z!github.com/bootjp/elastickv/protob\x06proto3" var ( file_distribution_proto_rawDescOnce sync.Once @@ -1269,24 +1767,32 @@ func file_distribution_proto_rawDescGZIP() []byte { return file_distribution_proto_rawDescData } -var file_distribution_proto_enumTypes = make([]protoimpl.EnumInfo, 4) -var file_distribution_proto_msgTypes = make([]protoimpl.MessageInfo, 11) +var file_distribution_proto_enumTypes = make([]protoimpl.EnumInfo, 5) +var file_distribution_proto_msgTypes = make([]protoimpl.MessageInfo, 18) 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 + (RouteState)(0), // 0: RouteState + (SplitJobPhase)(0), // 1: SplitJobPhase + (SplitJobBarrierState)(0), // 2: SplitJobBarrierState + (SplitJobExportPhase)(0), // 3: SplitJobExportPhase + (CatalogDeltaMutationOp)(0), // 4: CatalogDeltaMutationOp + (*GetRouteRequest)(nil), // 5: GetRouteRequest + (*GetRouteResponse)(nil), // 6: GetRouteResponse + (*GetTimestampRequest)(nil), // 7: GetTimestampRequest + (*GetTimestampResponse)(nil), // 8: GetTimestampResponse + (*RouteDescriptor)(nil), // 9: RouteDescriptor + (*SplitJobBracketProgress)(nil), // 10: SplitJobBracketProgress + (*SplitJob)(nil), // 11: SplitJob + (*ListRoutesRequest)(nil), // 12: ListRoutesRequest + (*ListRoutesResponse)(nil), // 13: ListRoutesResponse + (*SplitRangeRequest)(nil), // 14: SplitRangeRequest + (*SplitRangeResponse)(nil), // 15: SplitRangeResponse + (*CatalogCapabilitiesRequest)(nil), // 16: CatalogCapabilitiesRequest + (*CatalogCapabilitiesResponse)(nil), // 17: CatalogCapabilitiesResponse + (*CatalogWatchRequest)(nil), // 18: CatalogWatchRequest + (*CatalogDeltaMutation)(nil), // 19: CatalogDeltaMutation + (*CatalogDeltaRecord)(nil), // 20: CatalogDeltaRecord + (*CatalogSnapshotReset)(nil), // 21: CatalogSnapshotReset + (*CatalogWatchEvent)(nil), // 22: CatalogWatchEvent } var file_distribution_proto_depIdxs = []int32{ 0, // 0: RouteDescriptor.state:type_name -> RouteState @@ -1296,23 +1802,33 @@ var file_distribution_proto_depIdxs = []int32{ 1, // 4: SplitJob.abandon_from_phase:type_name -> SplitJobPhase 2, // 5: SplitJob.cutover_read_fence_state:type_name -> SplitJobBarrierState 2, // 6: SplitJob.target_staged_readiness_state:type_name -> SplitJobBarrierState - 9, // 7: SplitJob.bracket_progress:type_name -> SplitJobBracketProgress - 8, // 8: ListRoutesResponse.routes:type_name -> RouteDescriptor - 8, // 9: SplitRangeResponse.left:type_name -> RouteDescriptor - 8, // 10: SplitRangeResponse.right:type_name -> RouteDescriptor - 4, // 11: Distribution.GetRoute:input_type -> GetRouteRequest - 6, // 12: Distribution.GetTimestamp:input_type -> GetTimestampRequest - 11, // 13: Distribution.ListRoutes:input_type -> ListRoutesRequest - 13, // 14: Distribution.SplitRange:input_type -> SplitRangeRequest - 5, // 15: Distribution.GetRoute:output_type -> GetRouteResponse - 7, // 16: Distribution.GetTimestamp:output_type -> GetTimestampResponse - 12, // 17: Distribution.ListRoutes:output_type -> ListRoutesResponse - 14, // 18: Distribution.SplitRange:output_type -> SplitRangeResponse - 15, // [15:19] is the sub-list for method output_type - 11, // [11:15] is the sub-list for method input_type - 11, // [11:11] is the sub-list for extension type_name - 11, // [11:11] is the sub-list for extension extendee - 0, // [0:11] is the sub-list for field type_name + 10, // 7: SplitJob.bracket_progress:type_name -> SplitJobBracketProgress + 9, // 8: ListRoutesResponse.routes:type_name -> RouteDescriptor + 9, // 9: SplitRangeResponse.left:type_name -> RouteDescriptor + 9, // 10: SplitRangeResponse.right:type_name -> RouteDescriptor + 4, // 11: CatalogDeltaMutation.op:type_name -> CatalogDeltaMutationOp + 9, // 12: CatalogDeltaMutation.route:type_name -> RouteDescriptor + 19, // 13: CatalogDeltaRecord.mutations:type_name -> CatalogDeltaMutation + 9, // 14: CatalogSnapshotReset.routes:type_name -> RouteDescriptor + 21, // 15: CatalogWatchEvent.snapshot:type_name -> CatalogSnapshotReset + 20, // 16: CatalogWatchEvent.delta:type_name -> CatalogDeltaRecord + 5, // 17: Distribution.GetRoute:input_type -> GetRouteRequest + 7, // 18: Distribution.GetTimestamp:input_type -> GetTimestampRequest + 12, // 19: Distribution.ListRoutes:input_type -> ListRoutesRequest + 14, // 20: Distribution.SplitRange:input_type -> SplitRangeRequest + 16, // 21: Distribution.GetCatalogCapabilities:input_type -> CatalogCapabilitiesRequest + 18, // 22: Distribution.WatchCatalog:input_type -> CatalogWatchRequest + 6, // 23: Distribution.GetRoute:output_type -> GetRouteResponse + 8, // 24: Distribution.GetTimestamp:output_type -> GetTimestampResponse + 13, // 25: Distribution.ListRoutes:output_type -> ListRoutesResponse + 15, // 26: Distribution.SplitRange:output_type -> SplitRangeResponse + 17, // 27: Distribution.GetCatalogCapabilities:output_type -> CatalogCapabilitiesResponse + 22, // 28: Distribution.WatchCatalog:output_type -> CatalogWatchEvent + 23, // [23:29] is the sub-list for method output_type + 17, // [17:23] is the sub-list for method input_type + 17, // [17:17] is the sub-list for extension type_name + 17, // [17:17] is the sub-list for extension extendee + 0, // [0:17] is the sub-list for field type_name } func init() { file_distribution_proto_init() } @@ -1320,13 +1836,17 @@ func file_distribution_proto_init() { if File_distribution_proto != nil { return } + file_distribution_proto_msgTypes[17].OneofWrappers = []any{ + (*CatalogWatchEvent_Snapshot)(nil), + (*CatalogWatchEvent_Delta)(nil), + } type x struct{} out := protoimpl.TypeBuilder{ File: protoimpl.DescBuilder{ GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_distribution_proto_rawDesc), len(file_distribution_proto_rawDesc)), - NumEnums: 4, - NumMessages: 11, + NumEnums: 5, + NumMessages: 18, NumExtensions: 0, NumServices: 1, }, diff --git a/proto/distribution.proto b/proto/distribution.proto index f2199d909..a71a64565 100644 --- a/proto/distribution.proto +++ b/proto/distribution.proto @@ -7,6 +7,8 @@ service Distribution { rpc GetTimestamp (GetTimestampRequest) returns (GetTimestampResponse) {} rpc ListRoutes (ListRoutesRequest) returns (ListRoutesResponse) {} rpc SplitRange (SplitRangeRequest) returns (SplitRangeResponse) {} + rpc GetCatalogCapabilities (CatalogCapabilitiesRequest) returns (CatalogCapabilitiesResponse) {} + rpc WatchCatalog (CatalogWatchRequest) returns (stream CatalogWatchEvent) {} } message GetRouteRequest { @@ -137,3 +139,48 @@ message SplitRangeResponse { RouteDescriptor left = 2; RouteDescriptor right = 3; } + +message CatalogCapabilitiesRequest {} + +message CatalogCapabilitiesResponse { + repeated uint32 supported_protocol_versions = 1; + uint64 current_version = 2; + uint64 oldest_delta_version = 3; + uint32 max_batch_size = 4; +} + +message CatalogWatchRequest { + uint32 protocol_version = 1; + uint64 after_version = 2; + uint32 max_batch_size = 3; +} + +enum CatalogDeltaMutationOp { + CATALOG_DELTA_MUTATION_OP_UNSPECIFIED = 0; + CATALOG_DELTA_MUTATION_OP_UPSERT = 1; + CATALOG_DELTA_MUTATION_OP_DELETE = 2; +} + +message CatalogDeltaMutation { + CatalogDeltaMutationOp op = 1; + uint64 route_id = 2; + RouteDescriptor route = 3; +} + +message CatalogDeltaRecord { + uint64 previous_version = 1; + uint64 version = 2; + repeated CatalogDeltaMutation mutations = 3; +} + +message CatalogSnapshotReset { + uint64 version = 1; + repeated RouteDescriptor routes = 2; +} + +message CatalogWatchEvent { + oneof payload { + CatalogSnapshotReset snapshot = 1; + CatalogDeltaRecord delta = 2; + } +} diff --git a/proto/distribution_grpc.pb.go b/proto/distribution_grpc.pb.go index f8d9b82e4..610f7d6f2 100644 --- a/proto/distribution_grpc.pb.go +++ b/proto/distribution_grpc.pb.go @@ -19,10 +19,12 @@ 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_GetRoute_FullMethodName = "/Distribution/GetRoute" + Distribution_GetTimestamp_FullMethodName = "/Distribution/GetTimestamp" + Distribution_ListRoutes_FullMethodName = "/Distribution/ListRoutes" + Distribution_SplitRange_FullMethodName = "/Distribution/SplitRange" + Distribution_GetCatalogCapabilities_FullMethodName = "/Distribution/GetCatalogCapabilities" + Distribution_WatchCatalog_FullMethodName = "/Distribution/WatchCatalog" ) // DistributionClient is the client API for Distribution service. @@ -33,6 +35,8 @@ type DistributionClient interface { GetTimestamp(ctx context.Context, in *GetTimestampRequest, opts ...grpc.CallOption) (*GetTimestampResponse, error) ListRoutes(ctx context.Context, in *ListRoutesRequest, opts ...grpc.CallOption) (*ListRoutesResponse, error) SplitRange(ctx context.Context, in *SplitRangeRequest, opts ...grpc.CallOption) (*SplitRangeResponse, error) + GetCatalogCapabilities(ctx context.Context, in *CatalogCapabilitiesRequest, opts ...grpc.CallOption) (*CatalogCapabilitiesResponse, error) + WatchCatalog(ctx context.Context, in *CatalogWatchRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[CatalogWatchEvent], error) } type distributionClient struct { @@ -83,6 +87,35 @@ func (c *distributionClient) SplitRange(ctx context.Context, in *SplitRangeReque return out, nil } +func (c *distributionClient) GetCatalogCapabilities(ctx context.Context, in *CatalogCapabilitiesRequest, opts ...grpc.CallOption) (*CatalogCapabilitiesResponse, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(CatalogCapabilitiesResponse) + err := c.cc.Invoke(ctx, Distribution_GetCatalogCapabilities_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + +func (c *distributionClient) WatchCatalog(ctx context.Context, in *CatalogWatchRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[CatalogWatchEvent], error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + stream, err := c.cc.NewStream(ctx, &Distribution_ServiceDesc.Streams[0], Distribution_WatchCatalog_FullMethodName, cOpts...) + if err != nil { + return nil, err + } + x := &grpc.GenericClientStream[CatalogWatchRequest, CatalogWatchEvent]{ClientStream: stream} + if err := x.ClientStream.SendMsg(in); err != nil { + return nil, err + } + if err := x.ClientStream.CloseSend(); err != nil { + return nil, err + } + return x, nil +} + +// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. +type Distribution_WatchCatalogClient = grpc.ServerStreamingClient[CatalogWatchEvent] + // DistributionServer is the server API for Distribution service. // All implementations must embed UnimplementedDistributionServer // for forward compatibility. @@ -91,6 +124,8 @@ type DistributionServer interface { GetTimestamp(context.Context, *GetTimestampRequest) (*GetTimestampResponse, error) ListRoutes(context.Context, *ListRoutesRequest) (*ListRoutesResponse, error) SplitRange(context.Context, *SplitRangeRequest) (*SplitRangeResponse, error) + GetCatalogCapabilities(context.Context, *CatalogCapabilitiesRequest) (*CatalogCapabilitiesResponse, error) + WatchCatalog(*CatalogWatchRequest, grpc.ServerStreamingServer[CatalogWatchEvent]) error mustEmbedUnimplementedDistributionServer() } @@ -113,6 +148,12 @@ func (UnimplementedDistributionServer) ListRoutes(context.Context, *ListRoutesRe func (UnimplementedDistributionServer) SplitRange(context.Context, *SplitRangeRequest) (*SplitRangeResponse, error) { return nil, status.Error(codes.Unimplemented, "method SplitRange not implemented") } +func (UnimplementedDistributionServer) GetCatalogCapabilities(context.Context, *CatalogCapabilitiesRequest) (*CatalogCapabilitiesResponse, error) { + return nil, status.Error(codes.Unimplemented, "method GetCatalogCapabilities not implemented") +} +func (UnimplementedDistributionServer) WatchCatalog(*CatalogWatchRequest, grpc.ServerStreamingServer[CatalogWatchEvent]) error { + return status.Error(codes.Unimplemented, "method WatchCatalog not implemented") +} func (UnimplementedDistributionServer) mustEmbedUnimplementedDistributionServer() {} func (UnimplementedDistributionServer) testEmbeddedByValue() {} @@ -206,6 +247,35 @@ func _Distribution_SplitRange_Handler(srv interface{}, ctx context.Context, dec return interceptor(ctx, in, info, handler) } +func _Distribution_GetCatalogCapabilities_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(CatalogCapabilitiesRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(DistributionServer).GetCatalogCapabilities(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: Distribution_GetCatalogCapabilities_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(DistributionServer).GetCatalogCapabilities(ctx, req.(*CatalogCapabilitiesRequest)) + } + return interceptor(ctx, in, info, handler) +} + +func _Distribution_WatchCatalog_Handler(srv interface{}, stream grpc.ServerStream) error { + m := new(CatalogWatchRequest) + if err := stream.RecvMsg(m); err != nil { + return err + } + return srv.(DistributionServer).WatchCatalog(m, &grpc.GenericServerStream[CatalogWatchRequest, CatalogWatchEvent]{ServerStream: stream}) +} + +// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. +type Distribution_WatchCatalogServer = grpc.ServerStreamingServer[CatalogWatchEvent] + // Distribution_ServiceDesc is the grpc.ServiceDesc for Distribution service. // It's only intended for direct use with grpc.RegisterService, // and not to be introspected or modified (even as a copy) @@ -229,7 +299,17 @@ var Distribution_ServiceDesc = grpc.ServiceDesc{ MethodName: "SplitRange", Handler: _Distribution_SplitRange_Handler, }, + { + MethodName: "GetCatalogCapabilities", + Handler: _Distribution_GetCatalogCapabilities_Handler, + }, + }, + Streams: []grpc.StreamDesc{ + { + StreamName: "WatchCatalog", + Handler: _Distribution_WatchCatalog_Handler, + ServerStreams: true, + }, }, - Streams: []grpc.StreamDesc{}, Metadata: "distribution.proto", }