br: fix Log Backup unexpected paused when adding a already long-runni…

…ng task (pingcap#53695) close pingcap#53561
ghazalfamilyusa · Jun 11, 2024 · 64f5427 · 64f5427
1 parent 395edab
commit 64f5427
Show file tree

Hide file tree

Showing 5 changed files with 282 additions and 34 deletions.
diff --git a/br/pkg/streamhelper/BUILD.bazel b/br/pkg/streamhelper/BUILD.bazel
@@ -69,7 +69,7 @@ go_test(
     ],
     flaky = True,
     race = "on",
-    shard_count = 28,
+    shard_count = 32,
     deps = [
         ":streamhelper",
         "//br/pkg/errors",

diff --git a/br/pkg/streamhelper/advancer.go b/br/pkg/streamhelper/advancer.go
@@ -424,8 +424,17 @@ 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.setCheckpoints(spans.Sorted(spans.NewFullWith(e.Ranges, 0)))
-		c.lastCheckpoint = newCheckpointWithTS(e.Info.StartTs)
-		p, err := c.env.BlockGCUntil(ctx, c.task.StartTs)
+		globalCheckpointTs, err := c.env.GetGlobalCheckpointForTask(ctx, e.Name)
+		if err != nil {
+			log.Error("failed to get global checkpoint, skipping.", logutil.ShortError(err))
+			return err
+		}
+		if globalCheckpointTs < c.task.StartTs {
+			globalCheckpointTs = c.task.StartTs
+		}
+		log.Info("get global checkpoint", zap.Uint64("checkpoint", globalCheckpointTs))
+		c.lastCheckpoint = newCheckpointWithTS(globalCheckpointTs)
+		p, err := c.env.BlockGCUntil(ctx, globalCheckpointTs)
 		if err != nil {
 			log.Warn("failed to upload service GC safepoint, skipping.", logutil.ShortError(err))
 		}

diff --git a/br/pkg/streamhelper/advancer_env.go b/br/pkg/streamhelper/advancer_env.go
@@ -172,6 +172,8 @@ type StreamMeta interface {
 	Begin(ctx context.Context, ch chan<- TaskEvent) error
 	// UploadV3GlobalCheckpointForTask uploads the global checkpoint to the meta store.
 	UploadV3GlobalCheckpointForTask(ctx context.Context, taskName string, checkpoint uint64) error
+	// GetGlobalCheckpointForTask gets the global checkpoint from the meta store.
+	GetGlobalCheckpointForTask(ctx context.Context, taskName string) (uint64, error)
 	// ClearV3GlobalCheckpointForTask clears the global checkpoint to the meta store.
 	ClearV3GlobalCheckpointForTask(ctx context.Context, taskName string) error
 	PauseTask(ctx context.Context, taskName string) error