unistore: Implement the new PD API for GC

pkg/store/gcworker/BUILD.bazel

@@ -78,6 +78,7 @@ go_test(
         "@com_github_tikv_client_go_v2//tikvrpc",
         "@com_github_tikv_client_go_v2//txnkv/txnlock",
         "@com_github_tikv_pd_client//:client",
+        "@com_github_tikv_pd_client//constants",
         "@org_uber_go_goleak//:goleak",
     ],
 )

pkg/store/gcworker/gc_worker.go

@@ -694,6 +694,11 @@ func (w *GCWorker) calcNewSafePoint(ctx context.Context, now time.Time) (*time.T
 	safePointValue := w.calcSafePointByMinStartTS(ctx, oracle.GoTimeToTS(now.Add(-*lifeTime)))
 	safePointValue, err = w.setGCWorkerServiceSafePoint(ctx, safePointValue)
 	if err != nil {
+		// Temporary solution. This code should be refactored when errors returned by PD client can be typed.
+		if strings.Contains(err.Error(), "PD:gc:ErrDecreasingTxnSafePoint") {
+			logutil.BgLogger().Info("set gc worker service safe point is causing decreasing, the GC will be skipped", zap.Error(err))
+			return nil, 9, nil
+		}
 		return nil, 0, errors.Trace(err)
 	}
 

pkg/store/gcworker/gc_worker_test.go

@@ -48,6 +48,7 @@ import (
 	"github.com/tikv/client-go/v2/tikvrpc"
 	"github.com/tikv/client-go/v2/txnkv/txnlock"
 	pd "github.com/tikv/pd/client"
+	"github.com/tikv/pd/client/constants"
 )
 
 type mockGCWorkerLockResolver struct {
@@ -1290,7 +1291,12 @@ func TestRunGCJob(t *testing.T) {
 	useDistributedGC := s.gcWorker.checkUseDistributedGC()
 	require.True(t, useDistributedGC)
 	safePoint := s.mustAllocTs(t)
-	err := s.gcWorker.runGCJob(gcContext(), safePoint, gcConcurrency{1, false})
+	ctl := s.pdClient.GetGCInternalController(constants.NullKeyspaceID)
+	// runGCJob doesn't contain the AdvanceTxnSafePoint step. Do it explicitly.
+	res, err := ctl.AdvanceTxnSafePoint(gcContext(), safePoint)
+	require.NoError(t, err)
+	require.Equal(t, safePoint, res.NewTxnSafePoint)
+	err = s.gcWorker.runGCJob(gcContext(), safePoint, gcConcurrency{1, false})
 	require.NoError(t, err)
 
 	pdSafePoint := s.mustGetSafePointFromPd(t)
@@ -1316,6 +1322,9 @@ func TestRunGCJob(t *testing.T) {
 
 	p := s.createGCProbe(t, "k1")
 	safePoint = s.mustAllocTs(t)
+	res, err = ctl.AdvanceTxnSafePoint(gcContext(), safePoint)
+	require.NoError(t, err)
+	require.Equal(t, safePoint, res.NewTxnSafePoint)
 	err = s.gcWorker.runGCJob(gcContext(), safePoint, gcConcurrency{1, false})
 	require.NoError(t, err)
 	s.checkCollected(t, p)
@@ -1352,8 +1361,14 @@ func TestSetServiceSafePoint(t *testing.T) {
 	require.Equal(t, safePoint-10, s.mustGetMinServiceSafePointFromPd(t))
 
 	// Test removing the minimum service safe point.
-	s.mustRemoveServiceGCSafePoint(t, "svc1", safePoint-10, safePoint)
-	require.Equal(t, safePoint, s.mustGetMinServiceSafePointFromPd(t))
+	// As UpdateServiceGCSafePoint in unistore has become the compatible wrapper around GC barrier interface, this
+	// behavior has changed: the simulated service safe point for "gc_worker" will be blocked at `safePoint-10`.
+	// s.mustRemoveServiceGCSafePoint(t, "svc1", safePoint-10, safePoint)
+	// require.Equal(t, safePoint, s.mustGetMinServiceSafePointFromPd(t))
+	s.mustRemoveServiceGCSafePoint(t, "svc1", safePoint-10, safePoint-10)
+	require.Equal(t, safePoint-10, s.mustGetMinServiceSafePointFromPd(t))
+	// Advance it to `safePoint.
+	s.mustSetTiDBServiceSafePoint(t, safePoint, safePoint)
 
 	// Test the case when there are many safePoints.
 	safePoint += 100
@@ -1540,7 +1555,7 @@ func TestGCWithPendingTxn(t *testing.T) {
 	spkv := s.tikvStore.GetSafePointKV()
 	err = spkv.Put(fmt.Sprintf("%s/%s", infosync.ServerMinStartTSPath, "a"), strconv.FormatUint(txn.StartTS(), 10))
 	require.NoError(t, err)
-	s.mustSetTiDBServiceSafePoint(t, txn.StartTS(), txn.StartTS())
+	//s.mustSetTiDBServiceSafePoint(t, txn.StartTS(), txn.StartTS())
 	veryLong := gcDefaultLifeTime * 100
 	err = s.gcWorker.saveTime(gcLastRunTimeKey, oracle.GetTimeFromTS(s.mustAllocTs(t)).Add(-veryLong))
 	require.NoError(t, err)
@@ -1583,7 +1598,7 @@ func TestGCWithPendingTxn2(t *testing.T) {
 	spkv := s.tikvStore.GetSafePointKV()
 	err = spkv.Put(fmt.Sprintf("%s/%s", infosync.ServerMinStartTSPath, "a"), strconv.FormatUint(now, 10))
 	require.NoError(t, err)
-	s.mustSetTiDBServiceSafePoint(t, now, now)
+	//s.mustSetTiDBServiceSafePoint(t, now, now)
 	veryLong := gcDefaultLifeTime * 100
 	err = s.gcWorker.saveTime(gcLastRunTimeKey, oracle.GetTimeFromTS(s.mustAllocTs(t)).Add(-veryLong))
 	require.NoError(t, err)

pkg/store/mockstore/unistore/BUILD.bazel

@@ -41,7 +41,6 @@ go_library(
         "@com_github_tikv_client_go_v2//tikv",
         "@com_github_tikv_client_go_v2//tikvrpc",
         "@com_github_tikv_pd_client//:client",
-        "@com_github_tikv_pd_client//clients/gc",
         "@com_github_tikv_pd_client//clients/router",
         "@com_github_tikv_pd_client//clients/tso",
         "@com_github_tikv_pd_client//constants",

pkg/store/mockstore/unistore/pd.go

@@ -18,7 +18,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"math"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -32,7 +31,6 @@ import (
 	us "github.com/pingcap/tidb/pkg/store/mockstore/unistore/tikv"
 	"github.com/tikv/client-go/v2/oracle"
 	pd "github.com/tikv/pd/client"
-	"github.com/tikv/pd/client/clients/gc"
 	"github.com/tikv/pd/client/clients/router"
 	"github.com/tikv/pd/client/clients/tso"
 	"github.com/tikv/pd/client/constants"
@@ -48,8 +46,6 @@ type pdClient struct {
 	*us.MockPD
 	pd.ResourceManagerClient
 
-	serviceSafePoints map[string]uint64
-	gcSafePointMu     sync.Mutex
 	globalConfig      map[string]string
 	externalTimestamp atomic.Uint64
 
@@ -68,7 +64,6 @@ func newPDClient(pd *us.MockPD, addrs []string, keyspaceMeta *keyspacepb.Keyspac
 	return &pdClient{
 		MockPD:                pd,
 		ResourceManagerClient: infosync.NewMockResourceManagerClient(keyspaceID),
-		serviceSafePoints:     make(map[string]uint64),
 		globalConfig:          make(map[string]string),
 		addrs:                 addrs,
 		keyspaceMeta:          keyspaceMeta,
@@ -271,35 +266,6 @@ func (m *mockTSFuture) Wait() (int64, int64, error) {
 
 func (c *pdClient) GetLeaderURL() string { return "mockpd" }
 
-func (c *pdClient) UpdateServiceGCSafePoint(ctx context.Context, serviceID string, ttl int64, safePoint uint64) (uint64, error) {
-	c.gcSafePointMu.Lock()
-	defer c.gcSafePointMu.Unlock()
-
-	if ttl == 0 {
-		delete(c.serviceSafePoints, serviceID)
-	} else {
-		var minSafePoint uint64 = math.MaxUint64
-		for _, ssp := range c.serviceSafePoints {
-			if ssp < minSafePoint {
-				minSafePoint = ssp
-			}
-		}
-
-		if len(c.serviceSafePoints) == 0 || minSafePoint <= safePoint {
-			c.serviceSafePoints[serviceID] = safePoint
-		}
-	}
-
-	// The minSafePoint may have changed. Reload it.
-	var minSafePoint uint64 = math.MaxUint64
-	for _, ssp := range c.serviceSafePoints {
-		if ssp < minSafePoint {
-			minSafePoint = ssp
-		}
-	}
-	return minSafePoint, nil
-}
-
 func (c *pdClient) GetOperator(ctx context.Context, regionID uint64) (*pdpb.GetOperatorResponse, error) {
 	return &pdpb.GetOperatorResponse{Status: pdpb.OperatorStatus_SUCCESS}, nil
 }
@@ -431,11 +397,3 @@ func (c *pdClient) WatchGCSafePointV2(ctx context.Context, revision int64) (chan
 func (c *pdClient) WithCallerComponent(component caller.Component) pd.Client {
 	return c
 }
-
-func (c *pdClient) GetGCInternalController(keyspaceID uint32) gc.InternalController {
-	return nil
-}
-
-func (c *pdClient) GetGCStatesClient(keyspaceID uint32) gc.GCStatesClient {
-	return nil
-}

pkg/store/mockstore/unistore/tikv/BUILD.bazel

@@ -32,6 +32,7 @@ go_library(
         "//pkg/tablecodec",
         "//pkg/types",
         "//pkg/util/codec",
+        "//pkg/util/logutil",
         "//pkg/util/rowcodec",
         "@com_github_dgryski_go_farm//:go-farm",
         "@com_github_gogo_protobuf//proto",
@@ -51,7 +52,9 @@ go_library(
         "@com_github_pingcap_log//:log",
         "@com_github_pingcap_tipb//go-tipb",
         "@com_github_tikv_client_go_v2//oracle",
+        "@com_github_tikv_pd_client//clients/gc",
         "@com_github_tikv_pd_client//clients/router",
+        "@com_github_tikv_pd_client//constants",
         "@com_github_tikv_pd_client//opt",
         "@org_golang_google_grpc//:grpc",
         "@org_golang_google_grpc//credentials/insecure",
@@ -65,11 +68,12 @@ go_test(
     srcs = [
         "detector_test.go",
         "main_test.go",
+        "mock_pd_test.go",
         "mvcc_test.go",
     ],
     embed = [":tikv"],
     flaky = True,
-    shard_count = 28,
+    shard_count = 29,
     deps = [
         "//pkg/store/mockstore/unistore/config",
         "//pkg/store/mockstore/unistore/lockstore",
@@ -84,6 +88,8 @@ go_test(
         "@com_github_pingcap_kvproto//pkg/kvrpcpb",
         "@com_github_pingcap_kvproto//pkg/metapb",
         "@com_github_stretchr_testify//require",
+        "@com_github_tikv_pd_client//clients/gc",
+        "@com_github_tikv_pd_client//constants",
         "@org_uber_go_goleak//:goleak",
     ],
 )

pkg/store/mockstore/unistore/tikv/mock_pd_test.go

@@ -0,0 +1,122 @@
+// Copyright 2025 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package tikv
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	pdgc "github.com/tikv/pd/client/clients/gc"
+	"github.com/tikv/pd/client/constants"
+)
+
+func TestMockGCStatesManager(t *testing.T) {
+	re := require.New(t)
+	m := newGCStatesManager()
+	ctx := context.Background()
+
+	for _, keyspaceID := range []uint32{constants.NullKeyspaceID, 0, 1} {
+		ctl := m.GetGCInternalController(keyspaceID)
+		cli := m.GetGCStatesClient(keyspaceID)
+
+		s, err := cli.GetGCState(ctx)
+		re.NoError(err)
+		re.Equal(uint64(0), s.GCSafePoint)
+		re.Equal(uint64(0), s.TxnSafePoint)
+		re.Empty(s.GCBarriers)
+
+		tspRes, err := ctl.AdvanceTxnSafePoint(ctx, 10)
+		re.NoError(err)
+		re.Equal(uint64(10), tspRes.NewTxnSafePoint)
+		re.Equal(uint64(10), tspRes.Target)
+		re.Equal(uint64(0), tspRes.OldTxnSafePoint)
+		re.Empty(tspRes.BlockerDescription)
+
+		gspRes, err := ctl.AdvanceGCSafePoint(ctx, 9)
+		re.NoError(err)
+		re.Equal(uint64(9), gspRes.NewGCSafePoint)
+		re.Equal(uint64(9), gspRes.Target)
+		re.Equal(uint64(0), gspRes.OldGCSafePoint)
+
+		// No decreasing
+		_, err = ctl.AdvanceTxnSafePoint(ctx, 9)
+		re.Error(err)
+		_, err = ctl.AdvanceGCSafePoint(ctx, 8)
+		re.Error(err)
+		_, err = ctl.AdvanceGCSafePoint(ctx, 11)
+		re.Error(err)
+
+		s, err = cli.GetGCState(ctx)
+		re.NoError(err)
+		re.Equal(uint64(9), s.GCSafePoint)
+		re.Equal(uint64(10), s.TxnSafePoint)
+		re.Empty(s.GCBarriers)
+
+		// Invalid arguments
+		_, err = cli.SetGCBarrier(ctx, "", 20, pdgc.TTLNeverExpire)
+		re.Error(err)
+		_, err = cli.SetGCBarrier(ctx, "b1", 0, pdgc.TTLNeverExpire)
+		re.Error(err)
+		_, err = cli.SetGCBarrier(ctx, "b1", 20, 0)
+		re.Error(err)
+
+		// Disallow being less than txn safe point
+		_, err = cli.SetGCBarrier(ctx, "b1", 9, pdgc.TTLNeverExpire)
+		re.Error(err)
+
+		b, err := cli.SetGCBarrier(ctx, "b2", 25, pdgc.TTLNeverExpire)
+		re.NoError(err)
+		re.Equal("b2", b.BarrierID)
+		re.Equal(uint64(25), b.BarrierTS)
+
+		s, err = cli.GetGCState(ctx)
+		re.NoError(err)
+		re.Equal(uint64(9), s.GCSafePoint)
+		re.Equal(uint64(10), s.TxnSafePoint)
+		re.Len(s.GCBarriers, 1)
+		re.Equal("b2", s.GCBarriers[0].BarrierID)
+		re.Equal(uint64(25), s.GCBarriers[0].BarrierTS)
+
+		tspRes, err = ctl.AdvanceTxnSafePoint(ctx, 30)
+		re.NoError(err)
+		re.Equal(uint64(25), tspRes.NewTxnSafePoint)
+		re.Equal(uint64(30), tspRes.Target)
+		re.Equal(uint64(10), tspRes.OldTxnSafePoint)
+		re.Contains(tspRes.BlockerDescription, "b2")
+
+		s, err = cli.GetGCState(ctx)
+		re.NoError(err)
+		re.Equal(uint64(25), s.TxnSafePoint)
+
+		b, err = cli.DeleteGCBarrier(ctx, "b2")
+		re.NoError(err)
+		re.Equal("b2", b.BarrierID)
+		re.Equal(uint64(25), b.BarrierTS)
+
+		tspRes, err = ctl.AdvanceTxnSafePoint(ctx, 30)
+		re.NoError(err)
+		re.Equal(uint64(30), tspRes.NewTxnSafePoint)
+		re.Equal(uint64(30), tspRes.Target)
+		re.Equal(uint64(25), tspRes.OldTxnSafePoint)
+		re.Empty(tspRes.BlockerDescription)
+
+		s, err = cli.GetGCState(ctx)
+		re.NoError(err)
+		re.Equal(uint64(30), s.TxnSafePoint)
+		re.Equal(uint64(9), s.GCSafePoint)
+		re.Empty(s.GCBarriers)
+	}
+}