Log Backup: decouple log backup resolve locks from GCWorker.

br/pkg/conn/conn.go

@@ -245,6 +245,11 @@ func (mgr *Mgr) GetTLSConfig() *tls.Config {
 	return mgr.StoreManager.TLSConfig()
 }
 
+// GetStore gets the tikvStore.
+func (mgr *Mgr) GetStore() tikv.Storage {
+	return mgr.tikvStore
+}
+
 // GetLockResolver gets the LockResolver.
 func (mgr *Mgr) GetLockResolver() *txnlock.LockResolver {
 	return mgr.tikvStore.GetLockResolver()

br/pkg/streamhelper/advancer.go

@@ -17,7 +17,11 @@ import (
 	"github.com/pingcap/tidb/br/pkg/utils"
 	"github.com/pingcap/tidb/kv"
 	"github.com/pingcap/tidb/metrics"
+	"github.com/pingcap/tidb/util/codec"
+	"github.com/pingcap/tidb/util/gcutil"
+	tikvstore "github.com/tikv/client-go/v2/kv"
 	"github.com/tikv/client-go/v2/oracle"
+	"github.com/tikv/client-go/v2/txnkv/rangetask"
 	"go.uber.org/multierr"
 	"go.uber.org/zap"
 	"golang.org/x/sync/errgroup"
@@ -60,7 +64,7 @@ type CheckpointAdvancer struct {
 
 	// the cached last checkpoint.
 	// if no progress, this cache can help us don't to send useless requests.
-	lastCheckpoint uint64
+	lastCheckpoint checkpoint
 
 	checkpoints   *spans.ValueSortedFull
 	checkpointsMu sync.Mutex
@@ -69,6 +73,45 @@ type CheckpointAdvancer struct {
 	subscriberMu sync.Mutex
 }
 
+// checkpoint represents the TS with specific range.
+// it's only used in advancer.go.
+type checkpoint struct {
+	StartKey []byte
+	EndKey   []byte
+	TS       uint64
+
+	// It's better to use PD timestamp in future, for now
+	// use local time to decide the time to resolve lock is ok.
+	generateTime time.Time
+}
+
+func newCheckpointWithTS(ts uint64) checkpoint {
+	return checkpoint{
+		TS:           ts,
+		generateTime: time.Now(),
+	}
+}
+
+func newCheckpointWithSpan(s spans.Valued) checkpoint {
+	return checkpoint{
+		StartKey:     s.Key.StartKey,
+		EndKey:       s.Key.EndKey,
+		TS:           s.Value,
+		generateTime: time.Now(),
+	}
+}
+
+func (c checkpoint) safeTS() uint64 {
+	return c.TS - 1
+}
+
+// if a checkpoint stay in a time too long(1 min)
+// we should try to resolve lock for the range
+// to keep the RPO in a short value.
+func (c checkpoint) needResolveLocks() bool {
+	return time.Since(c.generateTime) > time.Minute
+}
+
 // NewCheckpointAdvancer creates a checkpoint advancer with the env.
 func NewCheckpointAdvancer(env Env) *CheckpointAdvancer {
 	return &CheckpointAdvancer{
@@ -180,7 +223,7 @@ func (c *CheckpointAdvancer) WithCheckpoints(f func(*spans.ValueSortedFull)) {
 }
 
 func (c *CheckpointAdvancer) CalculateGlobalCheckpointLight(ctx context.Context,
-	threshold time.Duration) (uint64, error) {
+	threshold time.Duration) (spans.Valued, error) {
 	var targets []spans.Valued
 	var minValue spans.Valued
 	c.WithCheckpoints(func(vsf *spans.ValueSortedFull) {
@@ -194,13 +237,13 @@ func (c *CheckpointAdvancer) CalculateGlobalCheckpointLight(ctx context.Context,
 		zap.Stringer("min", minValue), zap.Int("for-polling", len(targets)),
 		zap.String("min-ts", oracle.GetTimeFromTS(minValue.Value).Format(time.RFC3339)))
 	if len(targets) == 0 {
-		return minValue.Value, nil
+		return minValue, nil
 	}
 	err := c.tryAdvance(ctx, len(targets), func(i int) kv.KeyRange { return targets[i].Key })
 	if err != nil {
-		return 0, err
+		return minValue, err
 	}
-	return minValue.Value, nil
+	return minValue, nil
 }
 
 func (c *CheckpointAdvancer) consumeAllTask(ctx context.Context, ch <-chan TaskEvent) error {
@@ -293,7 +336,7 @@ func (c *CheckpointAdvancer) onTaskEvent(ctx context.Context, e TaskEvent) error
 		c.task = e.Info
 		c.taskRange = spans.Collapse(len(e.Ranges), func(i int) kv.KeyRange { return e.Ranges[i] })
 		c.checkpoints = spans.Sorted(spans.NewFullWith(e.Ranges, 0))
-		c.lastCheckpoint = e.Info.StartTs
+		c.lastCheckpoint = newCheckpointWithTS(e.Info.StartTs)
 		p, err := c.env.BlockGCUntil(ctx, c.task.StartTs)
 		if err != nil {
 			log.Warn("failed to upload service GC safepoint, skipping.", logutil.ShortError(err))
@@ -323,33 +366,57 @@ func (c *CheckpointAdvancer) onTaskEvent(ctx context.Context, e TaskEvent) error
 	return nil
 }
 
-func (c *CheckpointAdvancer) setCheckpoint(cp uint64) bool {
-	if cp < c.lastCheckpoint {
+func (c *CheckpointAdvancer) setCheckpoint(ctx context.Context, s spans.Valued) bool {
+	cp := newCheckpointWithSpan(s)
+	if cp.TS < c.lastCheckpoint.TS {
 		log.Warn("failed to update global checkpoint: stale",
-			zap.Uint64("old", c.lastCheckpoint), zap.Uint64("new", cp))
+			zap.Uint64("old", c.lastCheckpoint.TS), zap.Uint64("new", cp.TS))
+		return false
+	}
+	// lastCheckpoint is not increased too long enough.
+	// assume the cluster has expired locks for whatever reasons.
+	if c.lastCheckpoint.needResolveLocks() {
+		handler := func(ctx context.Context, r tikvstore.KeyRange) (rangetask.TaskStat, error) {
+			// we will scan all lock before cp.TS and try to resolve them by check txn status.
+			return gcutil.ResolveLocksForRange(ctx, "log backup advancer", c.env, cp.TS, r.StartKey, r.EndKey)
+		}
+		runner := rangetask.NewRangeTaskRunner("advancer-resolve-locks-runner", c.env.GetStore(), config.DefaultMaxConcurrencyAdvance, handler)
+		// Run resolve lock on the whole TiKV cluster. it will use startKey/endKey to scan region in PD. so we need encode key here.
+		encodedStartKey := codec.EncodeBytes([]byte{}, []byte(c.lastCheckpoint.StartKey))
+		encodedEndKey := codec.EncodeBytes([]byte{}, []byte(c.lastCheckpoint.EndKey))
+		err := runner.RunOnRange(ctx, encodedStartKey, encodedEndKey)
+		if err != nil {
+			log.Error("resolve locks failed", zap.String("category", "advancer"),
+				zap.String("uuid", "log backup advancer"),
+				zap.Error(err))
+			return false
+		}
+		log.Info("finish resolve locks", zap.String("category", "advancer"),
+			zap.String("uuid", "log backup advancer"),
+			zap.Int("regions", runner.CompletedRegions()))
 		return false
 	}
-	if cp <= c.lastCheckpoint {
+	if cp.TS <= c.lastCheckpoint.TS {
 		return false
 	}
 	c.lastCheckpoint = cp
-	metrics.LastCheckpoint.WithLabelValues(c.task.GetName()).Set(float64(c.lastCheckpoint))
+	metrics.LastCheckpoint.WithLabelValues(c.task.GetName()).Set(float64(c.lastCheckpoint.TS))
 	return true
 }
 
 // advanceCheckpointBy advances the checkpoint by a checkpoint getter function.
 func (c *CheckpointAdvancer) advanceCheckpointBy(ctx context.Context,
-	getCheckpoint func(context.Context) (uint64, error)) error {
+	getCheckpoint func(context.Context) (spans.Valued, error)) error {
 	start := time.Now()
 	cp, err := getCheckpoint(ctx)
 	if err != nil {
 		return err
 	}
 
-	if c.setCheckpoint(cp) {
+	if c.setCheckpoint(ctx, cp) {
 		log.Info("uploading checkpoint for task",
-			zap.Stringer("checkpoint", oracle.GetTimeFromTS(cp)),
-			zap.Uint64("checkpoint", cp),
+			zap.Stringer("checkpoint", oracle.GetTimeFromTS(cp.Value)),
+			zap.Uint64("checkpoint", cp.Value),
 			zap.String("task", c.task.Name),
 			zap.Stringer("take", time.Since(start)))
 	}
@@ -403,24 +470,24 @@ func (c *CheckpointAdvancer) subscribeTick(ctx context.Context) error {
 
 func (c *CheckpointAdvancer) importantTick(ctx context.Context) error {
 	c.checkpointsMu.Lock()
-	c.setCheckpoint(c.checkpoints.MinValue())
+	c.setCheckpoint(ctx, c.checkpoints.Min())
 	c.checkpointsMu.Unlock()
-	if err := c.env.UploadV3GlobalCheckpointForTask(ctx, c.task.Name, c.lastCheckpoint); err != nil {
+	if err := c.env.UploadV3GlobalCheckpointForTask(ctx, c.task.Name, c.lastCheckpoint.TS); err != nil {
 		return errors.Annotate(err, "failed to upload global checkpoint")
 	}
-	p, err := c.env.BlockGCUntil(ctx, c.lastCheckpoint-1)
+	p, err := c.env.BlockGCUntil(ctx, c.lastCheckpoint.safeTS())
 	if err != nil {
 		return errors.Annotatef(err,
 			"failed to update service GC safe point, current checkpoint is %d, target checkpoint is %d",
-			c.lastCheckpoint-1, p)
+			c.lastCheckpoint.safeTS(), p)
 	}
-	if p <= c.lastCheckpoint-1 {
+	if p <= c.lastCheckpoint.safeTS() {
 		log.Info("updated log backup GC safe point.",
-			zap.Uint64("checkpoint", p), zap.Uint64("target", c.lastCheckpoint-1))
+			zap.Uint64("checkpoint", p), zap.Uint64("target", c.lastCheckpoint.safeTS()))
 	}
-	if p > c.lastCheckpoint-1 {
+	if p > c.lastCheckpoint.safeTS() {
 		log.Warn("update log backup GC safe point failed: stale.",
-			zap.Uint64("checkpoint", p), zap.Uint64("target", c.lastCheckpoint-1))
+			zap.Uint64("checkpoint", p), zap.Uint64("target", c.lastCheckpoint.safeTS()))
 	}
 	return nil
 }
@@ -433,7 +500,7 @@ func (c *CheckpointAdvancer) optionalTick(cx context.Context) error {
 		threshold = c.Config().GetSubscriberErrorStartPollThreshold()
 	}
 
-	err := c.advanceCheckpointBy(cx, func(cx context.Context) (uint64, error) {
+	err := c.advanceCheckpointBy(cx, func(cx context.Context) (spans.Valued, error) {
 		return c.CalculateGlobalCheckpointLight(cx, threshold)
 	})
 	if err != nil {

br/pkg/streamhelper/advancer_env.go

@@ -6,10 +6,15 @@ import (
 	"context"
 	"time"
 
+	"github.com/pingcap/errors"
 	logbackup "github.com/pingcap/kvproto/pkg/logbackuppb"
 	"github.com/pingcap/tidb/br/pkg/utils"
 	"github.com/pingcap/tidb/config"
 	"github.com/pingcap/tidb/util/engine"
+	"github.com/pingcap/tidb/util/gcutil"
+	"github.com/tikv/client-go/v2/tikv"
+	"github.com/tikv/client-go/v2/tikvrpc"
+	"github.com/tikv/client-go/v2/txnkv/txnlock"
 	pd "github.com/tikv/pd/client"
 	clientv3 "go.etcd.io/etcd/client/v3"
 	"google.golang.org/grpc"
@@ -29,6 +34,8 @@ type Env interface {
 	LogBackupService
 	// StreamMeta connects to the metadata service (normally PD).
 	StreamMeta
+	// GCLockResolver try to resolve locks when region checkpoint stopped.
+	gcutil.GCLockResolver
 }
 
 // PDRegionScanner is a simple wrapper over PD
@@ -83,6 +90,7 @@ type clusterEnv struct {
 	clis *utils.StoreManager
 	*AdvancerExt
 	PDRegionScanner
+	*AdvancerLockResolver
 }
 
 // GetLogBackupClient gets the log backup client.
@@ -98,16 +106,17 @@ func (t clusterEnv) GetLogBackupClient(ctx context.Context, storeID uint64) (log
 }
 
 // CliEnv creates the Env for CLI usage.
-func CliEnv(cli *utils.StoreManager, etcdCli *clientv3.Client) Env {
+func CliEnv(cli *utils.StoreManager, tikvStore tikv.Storage, etcdCli *clientv3.Client) Env {
 	return clusterEnv{
-		clis:            cli,
-		AdvancerExt:     &AdvancerExt{MetaDataClient: *NewMetaDataClient(etcdCli)},
-		PDRegionScanner: PDRegionScanner{cli.PDClient()},
+		clis:                 cli,
+		AdvancerExt:          &AdvancerExt{MetaDataClient: *NewMetaDataClient(etcdCli)},
+		PDRegionScanner:      PDRegionScanner{cli.PDClient()},
+		AdvancerLockResolver: &AdvancerLockResolver{TiKvStore: tikvStore},
 	}
 }
 
 // TiDBEnv creates the Env by TiDB config.
-func TiDBEnv(pdCli pd.Client, etcdCli *clientv3.Client, conf *config.Config) (Env, error) {
+func TiDBEnv(tikvStore tikv.Storage, pdCli pd.Client, etcdCli *clientv3.Client, conf *config.Config) (Env, error) {
 	tconf, err := conf.GetTiKVConfig().Security.ToTLSConfig()
 	if err != nil {
 		return nil, err
@@ -117,8 +126,9 @@ func TiDBEnv(pdCli pd.Client, etcdCli *clientv3.Client, conf *config.Config) (En
 			Time:    time.Duration(conf.TiKVClient.GrpcKeepAliveTime) * time.Second,
 			Timeout: time.Duration(conf.TiKVClient.GrpcKeepAliveTimeout) * time.Second,
 		}, tconf),
-		AdvancerExt:     &AdvancerExt{MetaDataClient: *NewMetaDataClient(etcdCli)},
-		PDRegionScanner: PDRegionScanner{Client: pdCli},
+		AdvancerExt:          &AdvancerExt{MetaDataClient: *NewMetaDataClient(etcdCli)},
+		PDRegionScanner:      PDRegionScanner{Client: pdCli},
+		AdvancerLockResolver: &AdvancerLockResolver{TiKvStore: tikvStore},
 	}, nil
 }
 
@@ -137,3 +147,34 @@ type StreamMeta interface {
 	// ClearV3GlobalCheckpointForTask clears the global checkpoint to the meta store.
 	ClearV3GlobalCheckpointForTask(ctx context.Context, taskName string) error
 }
+
+type AdvancerLockResolver struct {
+	TiKvStore tikv.Storage
+}
+
+func (w *AdvancerLockResolver) LocateKey(bo *tikv.Backoffer, key []byte) (*tikv.KeyLocation, error) {
+	return w.TiKvStore.GetRegionCache().LocateKey(bo, key)
+}
+
+// ResolveLocks tries to resolve expired locks with this method.
+// It will check status of the txn. Resolve the lock if txn is expired, Or do nothing.
+func (w *AdvancerLockResolver) ResolveLocks(bo *tikv.Backoffer, locks []*txnlock.Lock, loc tikv.RegionVerID) (bool, error) {
+	if len(locks) == 0 {
+		return true, nil
+	}
+
+	_, err := w.TiKvStore.GetLockResolver().ResolveLocks(bo, 0, locks)
+	return err == nil, errors.Trace(err)
+}
+
+func (w *AdvancerLockResolver) ScanLocks(key []byte, regionID uint64) []*txnlock.Lock {
+	return nil
+}
+
+func (w *AdvancerLockResolver) SendReq(bo *tikv.Backoffer, req *tikvrpc.Request, regionID tikv.RegionVerID, timeout time.Duration) (*tikvrpc.Response, error) {
+	return w.TiKvStore.SendReq(bo, req, regionID, timeout)
+}
+
+func (w *AdvancerLockResolver) GetStore() tikv.Storage {
+	return w.TiKvStore
+}

br/pkg/task/stream.go

@@ -876,7 +876,7 @@ func RunStreamAdvancer(c context.Context, g glue.Glue, cmdName string, cfg *Stre
 	if err != nil {
 		return err
 	}
-	env := streamhelper.CliEnv(mgr.StoreManager, etcdCLI)
+	env := streamhelper.CliEnv(mgr.StoreManager, mgr.GetStore(), etcdCLI)
 	advancer := streamhelper.NewCheckpointAdvancer(env)
 	advancer.UpdateConfig(cfg.AdvancerCfg)
 	advancerd := daemon.New(advancer, streamhelper.OwnerManagerForLogBackup(ctx, etcdCLI), cfg.AdvancerCfg.TickDuration)

domain/domain.go

@@ -1274,7 +1274,12 @@ func (do *Domain) initLogBackup(ctx context.Context, pdClient pd.Client) error {
 		log.Warn("pd / etcd client not provided, won't begin Advancer.")
 		return nil
 	}
-	env, err := streamhelper.TiDBEnv(pdClient, do.etcdClient, cfg)
+	tikvStore, ok := do.Store().(tikv.Storage)
+	if !ok {
+		log.Warn("non tikv store, stop begin Advancer.")
+		return nil
+	}
+	env, err := streamhelper.TiDBEnv(tikvStore, pdClient, do.etcdClient, cfg)
 	if err != nil {
 		return err
 	}