server: Add gRPC interfaces for the new GC APIs.

client/go.mod

@@ -43,3 +43,5 @@ require (
 	gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
+
+replace github.com/pingcap/kvproto => github.com/MyonKeminta/kvproto v0.0.0-20250422131702-6653a56d7ef8

client/go.sum

@@ -1,5 +1,7 @@
 github.com/BurntSushi/toml v0.3.1 h1:WXkYYl6Yr3qBf1K79EBnL4mak0OimBfB0XUf9Vl28OQ=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
+github.com/MyonKeminta/kvproto v0.0.0-20250422131702-6653a56d7ef8 h1:lWRC7jXKjT9jlOf3isCsMHNMuvfOeQNLDl56bJ/dq+Q=
+github.com/MyonKeminta/kvproto v0.0.0-20250422131702-6653a56d7ef8/go.mod h1:rXxWk2UnwfUhLXha1jxRWPADw9eMZGWEWCg92Tgmb/8=
 github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=
 github.com/benbjohnson/clock v1.3.0 h1:ip6w0uFQkncKQ979AypyG0ER7mqUSBdKLOgAle/AT8A=
 github.com/benbjohnson/clock v1.3.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=
@@ -49,8 +51,6 @@ github.com/pingcap/errors v0.11.5-0.20211224045212-9687c2b0f87c h1:xpW9bvK+HuuTm
 github.com/pingcap/errors v0.11.5-0.20211224045212-9687c2b0f87c/go.mod h1:X2r9ueLEUZgtx2cIogM0v4Zj5uvvzhuuiu7Pn8HzMPg=
 github.com/pingcap/failpoint v0.0.0-20240528011301-b51a646c7c86 h1:tdMsjOqUR7YXHoBitzdebTvOjs/swniBTOLy5XiMtuE=
 github.com/pingcap/failpoint v0.0.0-20240528011301-b51a646c7c86/go.mod h1:exzhVYca3WRtd6gclGNErRWb1qEgff3LYta0LvRmON4=
-github.com/pingcap/kvproto v0.0.0-20250224053625-b6a98c6bf02d h1:52qhTQG8G8V/pHo/w7F4d2Tw98KMk2C+gAe3U8SWRAg=
-github.com/pingcap/kvproto v0.0.0-20250224053625-b6a98c6bf02d/go.mod h1:rXxWk2UnwfUhLXha1jxRWPADw9eMZGWEWCg92Tgmb/8=
 github.com/pingcap/log v1.1.1-0.20221110025148-ca232912c9f3 h1:HR/ylkkLmGdSSDaD8IDP+SZrdhV1Kibl9KrHxJ9eciw=
 github.com/pingcap/log v1.1.1-0.20221110025148-ca232912c9f3/go.mod h1:DWQW5jICDR7UJh4HtxXSM20Churx4CQL0fwL/SoOSA4=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=

go.mod

@@ -222,3 +222,5 @@ require (
 	moul.io/zapgorm2 v1.1.0 // indirect
 	sigs.k8s.io/yaml v1.4.0 // indirect
 )
+
+replace github.com/pingcap/kvproto => github.com/MyonKeminta/kvproto v0.0.0-20250422131702-6653a56d7ef8

pkg/gc/gc_state_manager.go

@@ -432,11 +432,14 @@ func (*GCStateManager) logAdvancingTxnSafePoint(keyspaceID uint32, result Advanc
 // GC barrier can extend its lifetime arbitrarily.
 //
 // Passing non-positive value to ttl is not allowed. Passing `time.Duration(math.MaxInt64)` to ttl indicates that the
-// GC barrier should never expire.
+// GC barrier should never expire. The ttl might be rounded up, and the actual ttl is guaranteed no less than the
+// specified duration.
 //
 // The barrierID must be non-empty. For NullKeyspace, "gc_worker" is a reserved name and cannot be used as a barrierID.
 //
 // The given barrierTS must be greater than or equal to the current txn safe point, or an error will be returned.
+//
+// When this function executes successfully, its result is never nil.
 func (m *GCStateManager) SetGCBarrier(keyspaceID uint32, barrierID string, barrierTS uint64, ttl time.Duration, now time.Time) (*endpoint.GCBarrier, error) {
 	if ttl <= 0 {
 		return nil, errs.ErrInvalidArgument.GenWithStackByArgs("ttl", ttl)

pkg/storage/endpoint/gc_states.go

@@ -165,8 +165,11 @@ func gcBarrierFromServiceSafePoint(s *ServiceSafePoint) *GCBarrier {
 	return res
 }
 
-// toServiceSafePoint converts the GCBarrier to a synonymous ServiceSafePoint for storing physically.
-func (b *GCBarrier) toServiceSafePoint(keyspaceID uint32) *ServiceSafePoint {
+// ToServiceSafePoint converts the GCBarrier to a synonymous ServiceSafePoint for storing physically.
+// This method should never be used unless handling the physical data, which needs to be compatible with service safe
+// points.
+// This method is public only for keeping some old HTTP API compatible.
+func (b *GCBarrier) ToServiceSafePoint(keyspaceID uint32) *ServiceSafePoint {
 	res := &ServiceSafePoint{
 		ServiceID:  b.BarrierID,
 		ExpiredAt:  math.MaxInt64,
@@ -489,7 +492,7 @@ func (wb *GCStateWriteBatch) SetTxnSafePoint(keyspaceID uint32, txnSafePoint uin
 // SetGCBarrier sets a GCBarrier with the given barrierID for a specific keyspace.
 func (wb *GCStateWriteBatch) SetGCBarrier(keyspaceID uint32, newGCBarrier *GCBarrier) error {
 	key := keypath.GCBarrierPath(keyspaceID, newGCBarrier.BarrierID)
-	return wb.writeJSON(key, newGCBarrier.toServiceSafePoint(keyspaceID))
+	return wb.writeJSON(key, newGCBarrier.ToServiceSafePoint(keyspaceID))
 }
 
 // DeleteGCBarrier deletes the GCBarrier with the given barrierID from a specific keyspace.

pkg/storage/endpoint/gc_states_test.go

@@ -80,7 +80,7 @@ func TestGCBarriersConversions(t *testing.T) {
 	// Test representing GC barriers by service safe points.
 	for i, gcBarrier := range gcBarriers {
 		expectedServiceSafePoint := serviceSafePoints[i]
-		serviceSafePoint := gcBarrier.toServiceSafePoint(keyspaces[i])
+		serviceSafePoint := gcBarrier.ToServiceSafePoint(keyspaces[i])
 		re.Equal(expectedServiceSafePoint, serviceSafePoint)
 	}
 

server/gc_service.go

@@ -19,6 +19,7 @@
 	"encoding/json"
 	"fmt"
 	"math"
+	"time"
 
 	clientv3 "go.etcd.io/etcd/client/v3"
 	"go.uber.org/zap"
@@ -28,10 +29,13 @@
 	"github.com/pingcap/log"
 
 	"github.com/tikv/pd/pkg/errs"
+	"github.com/tikv/pd/pkg/gc"
+	"github.com/tikv/pd/pkg/mcs/utils/constant"
 	"github.com/tikv/pd/pkg/storage/endpoint"
 	"github.com/tikv/pd/pkg/utils/etcdutil"
 	"github.com/tikv/pd/pkg/utils/keypath"
 	"github.com/tikv/pd/pkg/utils/tsoutil"
+	"github.com/tikv/pd/pkg/utils/typeutil"
 )
 
 // GetGCSafePointV2 return gc safe point for the given keyspace.
@@ -254,3 +258,280 @@
 	}
 	return values, resp.Header.Revision, nil
 }
+
+func getKeyspaceID(keyspaceScope *pdpb.KeyspaceScope) uint32 {
+	if keyspaceScope == nil {
+		return constant.NullKeyspaceID
+	}
+	return keyspaceScope.GetKeyspaceId()
+}
+
+func gcBarrierToProto(b *endpoint.GCBarrier, now time.Time) *pdpb.GCBarrierInfo {
+	if b == nil {
+		return nil
+	}
+
+	// After rounding, the actual TTL might be not exactly the same as the specified value. Recalculate it anyway.
+	// MaxInt64 represents that the expiration time is not specified and it never expires.
+	var resultTTL int64 = math.MaxInt64
+	if b.ExpirationTime != nil {
+		resultTTL = int64(math.Floor(b.ExpirationTime.Sub(now).Seconds()))
+		if resultTTL < 0 {
+			resultTTL = 0
+		}
+	}
+
+	return &pdpb.GCBarrierInfo{
+		BarrierId:  b.BarrierID,
+		BarrierTs:  b.BarrierTS,
+		TtlSeconds: resultTTL,
+	}
+}
+
+func gcStateToProto(gcState gc.GCState, now time.Time) *pdpb.GCState {
+	gcBarriers := make([]*pdpb.GCBarrierInfo, 0, len(gcState.GCBarriers))
+	for _, b := range gcState.GCBarriers {
+		gcBarriers = append(gcBarriers, gcBarrierToProto(b, now))
+	}
+	return &pdpb.GCState{
+		KeyspaceScope: &pdpb.KeyspaceScope{
+			KeyspaceId: gcState.KeyspaceID,
+		},
+		IsKeyspaceLevelGc: gcState.IsKeyspaceLevel,
+		TxnSafePoint:      gcState.TxnSafePoint,
+		GcSafePoint:       gcState.GCSafePoint,
+		GcBarriers:        gcBarriers,
+	}
+}
+
+// AdvanceGCSafePoint tries to advance the GC safe point.
+func (s *GrpcServer) AdvanceGCSafePoint(ctx context.Context, request *pdpb.AdvanceGCSafePointRequest) (*pdpb.AdvanceGCSafePointResponse, error) {
+	done, err := s.rateLimitCheck()
+	if err != nil {
+		return nil, err
+	}
+	if done != nil {
+		defer done()
+	}
+	fn := func(ctx context.Context, client *grpc.ClientConn) (any, error) {
+		return pdpb.NewPDClient(client).AdvanceGCSafePoint(ctx, request)
+	}
+	if rsp, err := s.unaryMiddleware(ctx, request, fn); err != nil {
+		return nil, err
+	} else if rsp != nil {
+		return rsp.(*pdpb.AdvanceGCSafePointResponse), err
+	}
+
+	rc := s.GetRaftCluster()
+	if rc == nil {
+		return &pdpb.AdvanceGCSafePointResponse{Header: notBootstrappedHeader()}, nil
+	}
+	oldGCSafePoint, newGCSafePoint, err := s.gcStateManager.AdvanceGCSafePoint(getKeyspaceID(request.GetKeyspaceScope()), request.GetTarget())
+	if err != nil {
+		return &pdpb.AdvanceGCSafePointResponse{
+			Header: wrapErrorToHeader(pdpb.ErrorType_UNKNOWN, err.Error()),
+		}, nil
+	}
+
+	return &pdpb.AdvanceGCSafePointResponse{
+		Header:         wrapHeader(),
+		OldGcSafePoint: oldGCSafePoint,
+		NewGcSafePoint: newGCSafePoint,
+	}, nil
+}
+
+// AdvanceTxnSafePoint tries to advance the transaction safe point.
+func (s *GrpcServer) AdvanceTxnSafePoint(ctx context.Context, request *pdpb.AdvanceTxnSafePointRequest) (*pdpb.AdvanceTxnSafePointResponse, error) {
+	done, err := s.rateLimitCheck()
+	if err != nil {
+		return nil, err
+	}
+	if done != nil {
+		defer done()
+	}
+	fn := func(ctx context.Context, client *grpc.ClientConn) (any, error) {
+		return pdpb.NewPDClient(client).AdvanceTxnSafePoint(ctx, request)
+	}
+	if rsp, err := s.unaryMiddleware(ctx, request, fn); err != nil {
+		return nil, err
+	} else if rsp != nil {
+		return rsp.(*pdpb.AdvanceTxnSafePointResponse), err
+	}
+
+	rc := s.GetRaftCluster()
+	if rc == nil {
+		return &pdpb.AdvanceTxnSafePointResponse{Header: notBootstrappedHeader()}, nil
+	}
+
+	res, err := s.gcStateManager.AdvanceTxnSafePoint(getKeyspaceID(request.GetKeyspaceScope()), request.GetTarget(), time.Now())
+	if err != nil {
+		return &pdpb.AdvanceTxnSafePointResponse{
+			Header: wrapErrorToHeader(pdpb.ErrorType_UNKNOWN, err.Error()),
+		}, nil
+	}
+
+	return &pdpb.AdvanceTxnSafePointResponse{
+		Header:             wrapHeader(),
+		OldTxnSafePoint:    res.OldTxnSafePoint,
+		NewTxnSafePoint:    res.NewTxnSafePoint,
+		BlockerDescription: res.BlockerDescription,
+	}, nil
+}
+
+// SetGCBarrier sets a GC barrier.
+func (s *GrpcServer) SetGCBarrier(ctx context.Context, request *pdpb.SetGCBarrierRequest) (*pdpb.SetGCBarrierResponse, error) {
+	done, err := s.rateLimitCheck()
+	if err != nil {
+		return nil, err
+	}
+	if done != nil {
+		defer done()
+	}
+	fn := func(ctx context.Context, client *grpc.ClientConn) (any, error) {
+		return pdpb.NewPDClient(client).SetGCBarrier(ctx, request)
+	}
+	if rsp, err := s.unaryMiddleware(ctx, request, fn); err != nil {
+		return nil, err
+	} else if rsp != nil {
+		return rsp.(*pdpb.SetGCBarrierResponse), err
+	}
+
+	rc := s.GetRaftCluster()
+	if rc == nil {
+		return &pdpb.SetGCBarrierResponse{Header: notBootstrappedHeader()}, nil
+	}
+
+	now := time.Now()
+	newBarrier, err := s.gcStateManager.SetGCBarrier(
+		getKeyspaceID(request.GetKeyspaceScope()),
+		request.GetBarrierId(),
+		request.GetBarrierTs(),
+		typeutil.SaturatingStdDurationFromSeconds(request.GetTtlSeconds()),
+		now)
+	if err != nil {
+		return &pdpb.SetGCBarrierResponse{
+			Header: wrapErrorToHeader(pdpb.ErrorType_UNKNOWN, err.Error()),
+		}, nil
+	}
+
+	return &pdpb.SetGCBarrierResponse{
+		Header:         wrapHeader(),
+		NewBarrierInfo: gcBarrierToProto(newBarrier, now),
+	}, nil
+}
+
+// DeleteGCBarrier deletes a GC barrier.
+func (s *GrpcServer) DeleteGCBarrier(ctx context.Context, request *pdpb.DeleteGCBarrierRequest) (*pdpb.DeleteGCBarrierResponse, error) {
+	done, err := s.rateLimitCheck()
+	if err != nil {
+		return nil, err
+	}
+	if done != nil {
+		defer done()
+	}
+	fn := func(ctx context.Context, client *grpc.ClientConn) (any, error) {
+		return pdpb.NewPDClient(client).DeleteGCBarrier(ctx, request)
+	}
+	if rsp, err := s.unaryMiddleware(ctx, request, fn); err != nil {
+		return nil, err
+	} else if rsp != nil {
+		return rsp.(*pdpb.DeleteGCBarrierResponse), err
+	}
+
+	rc := s.GetRaftCluster()
+	if rc == nil {
+		return &pdpb.DeleteGCBarrierResponse{Header: notBootstrappedHeader()}, nil
+	}
+
+	now := time.Now()
+
+	deletedBarrier, err := s.gcStateManager.DeleteGCBarrier(getKeyspaceID(request.GetKeyspaceScope()), request.GetBarrierId())
+	if err != nil {
+		return &pdpb.DeleteGCBarrierResponse{
+			Header: wrapErrorToHeader(pdpb.ErrorType_UNKNOWN, err.Error()),
+		}, nil
+	}
+
+	return &pdpb.DeleteGCBarrierResponse{
+		Header:             wrapHeader(),
+		DeletedBarrierInfo: gcBarrierToProto(deletedBarrier, now),
+	}, nil
+}
+
+// GetGCState gets the GC state.
+func (s *GrpcServer) GetGCState(ctx context.Context, request *pdpb.GetGCStateRequest) (*pdpb.GetGCStateResponse, error) {
+	done, err := s.rateLimitCheck()
+	if err != nil {
+		return nil, err
+	}
+	if done != nil {
+		defer done()
+	}
+	fn := func(ctx context.Context, client *grpc.ClientConn) (any, error) {
+		return pdpb.NewPDClient(client).GetGCState(ctx, request)
+	}
+	if rsp, err := s.unaryMiddleware(ctx, request, fn); err != nil {
+		return nil, err
+	} else if rsp != nil {
+		return rsp.(*pdpb.GetGCStateResponse), err
+	}
+
+	rc := s.GetRaftCluster()
+	if rc == nil {
+		return &pdpb.GetGCStateResponse{Header: notBootstrappedHeader()}, nil
+	}
+
+	gcState, err := s.gcStateManager.GetGCState(getKeyspaceID(request.GetKeyspaceScope()))
+	if err != nil {
+		return &pdpb.GetGCStateResponse{
+			Header: wrapErrorToHeader(pdpb.ErrorType_UNKNOWN, err.Error()),
+		}, nil
+	}
+
+	return &pdpb.GetGCStateResponse{
+		Header:  wrapHeader(),
+		GcState: gcStateToProto(gcState, time.Now()),
+	}, nil
+}
+
+// GetAllKeyspacesGCStates gets the GC states of all keyspaces.
+func (s *GrpcServer) GetAllKeyspacesGCStates(ctx context.Context, request *pdpb.GetAllKeyspacesGCStatesRequest) (*pdpb.GetAllKeyspacesGCStatesResponse, error) {
+	done, err := s.rateLimitCheck()
+	if err != nil {
+		return nil, err
+	}
+	if done != nil {
+		defer done()
+	}
+	fn := func(ctx context.Context, client *grpc.ClientConn) (any, error) {
+		return pdpb.NewPDClient(client).GetAllKeyspacesGCStates(ctx, request)
+	}
+	if rsp, err := s.unaryMiddleware(ctx, request, fn); err != nil {
+		return nil, err
+	} else if rsp != nil {
+		return rsp.(*pdpb.GetAllKeyspacesGCStatesResponse), err
+	}
+
+	rc := s.GetRaftCluster()
+	if rc == nil {
+		return &pdpb.GetAllKeyspacesGCStatesResponse{Header: notBootstrappedHeader()}, nil
+	}
+
+	gcStates, err := s.gcStateManager.GetAllKeyspacesGCStates()
+	if err != nil {
+		return &pdpb.GetAllKeyspacesGCStatesResponse{
+			Header: wrapErrorToHeader(pdpb.ErrorType_UNKNOWN, err.Error()),
+		}, nil
+	}
+
+	now := time.Now()
+	gcStatesPb := make([]*pdpb.GCState, 0, len(gcStates))
+	for _, gcState := range gcStates {
+		gcStatesPb = append(gcStatesPb, gcStateToProto(gcState, now))
+	}
+
+	return &pdpb.GetAllKeyspacesGCStatesResponse{
+		Header:   wrapHeader(),
+		GcStates: gcStatesPb,
+	}, nil
+}