set checkpoint to global checkpoint once start

Signed-off-by: Yu Juncen <yujuncen@pingcap.com>

components/backup-stream/src/async_task_manager.rs

@@ -283,14 +283,15 @@ where
                 metrics::INITIAL_SCAN_REASON
                     .with_label_values(&["region-changed"])
                     .inc();
-                if let Err(e) = self
-                    .observe_over_with_initial_data_from_checkpoint(
+                let r = async {
+                    Result::Ok(self.observe_over_with_initial_data_from_checkpoint(
                         region,
-                        for_task,
+                        self.get_last_checkpoint_of(&for_task).await?,
                         handle.clone(),
-                    )
-                    .await
-                {
+                    ))
+                }
+                .await;
+                if let Err(e) = r {
                     try_send!(
                         self.scheduler,
                         Task::ModifyObserve(ObserveOp::NotifyFailToStartObserve {
@@ -304,9 +305,8 @@ where
         }
     }
 
-    async fn start_observe(&self, region: Region) {
-        let handle = ObserveHandle::new();
-        let result = match self.find_task_by_region(&region) {
+    async fn try_start_observe(&self, region: &Region, handle: ObserveHandle) -> Result<()> {
+        match self.find_task_by_region(&region) {
             None => {
                 warn!(
                     "the region {:?} is register to no task but being observed (start_key = {}; end_key = {}; task_stat = {:?}): maybe stale, aborting",
@@ -315,20 +315,30 @@ where
                     utils::redact(&region.get_end_key()),
                     self.range_router
                 );
-                return;
             }
 
             Some(for_task) => {
-                self.observe_over_with_initial_data_from_checkpoint(
-                    &region,
-                    for_task,
-                    handle.clone(),
-                )
-                .await
+                let tso = self.get_last_checkpoint_of(&for_task).await?;
+                // We should set the local checkpoint to the newly added task ASAP,
+                // or once the original leader store of this region get rid of the slow-progress region,
+                // it may advance the global checkpoint unexpectlly.
+                // NOTE: maybe `spawn` here?
+                self.get_meta_client()
+                    .set_local_task_checkpoint(&for_task, tso.into_inner())
+                    .await?;
+                self.observe_over_with_initial_data_from_checkpoint(&region, tso, handle.clone());
             }
-        };
+        }
+        Ok(())
+    }
 
-        if let Err(err) = result {
+    fn get_meta_client(&self) -> &MetadataClient<S> {
+        self.meta_cli.as_ref().unwrap()
+    }
+
+    async fn start_observe(&self, region: Region) {
+        let handle = ObserveHandle::new();
+        if let Err(err) = self.try_start_observe(&region, handle.clone()).await {
             try_send!(
                 self.scheduler,
                 Task::ModifyObserve(ObserveOp::NotifyFailToStartObserve {
@@ -391,14 +401,18 @@ where
         Ok(())
     }
 
-    async fn observe_over_with_initial_data_from_checkpoint(
+    async fn get_last_checkpoint_of(&self, task: &str) -> Result<TimeStamp> {
+        let meta_cli = self.meta_cli.as_ref().unwrap().clone();
+        let last_checkpoint = TimeStamp::new(meta_cli.global_progress_of_task(task).await?);
+        Ok(last_checkpoint)
+    }
+
+    fn observe_over_with_initial_data_from_checkpoint(
         &self,
         region: &Region,
-        task: String,
+        last_checkpoint: TimeStamp,
         handle: ObserveHandle,
-    ) -> Result<()> {
-        let meta_cli = self.meta_cli.as_ref().unwrap().clone();
-        let last_checkpoint = TimeStamp::new(meta_cli.global_progress_of_task(&task).await?);
+    ) {
         self.subs
             .register_region(region, handle.clone(), Some(last_checkpoint));
 
@@ -417,7 +431,6 @@ where
                 region.get_id()
             ));
         }
-        Ok(())
     }
 
     fn find_task_by_region(&self, r: &Region) -> Option<String> {

components/backup-stream/src/endpoint.rs

@@ -1,9 +1,6 @@
 // Copyright 2022 TiKV Project Authors. Licensed under Apache-2.0.
 
-use std::{
-    convert::AsRef, fmt, marker::PhantomData, path::PathBuf, sync::Arc,
-    time::Duration,
-};
+use std::{convert::AsRef, fmt, marker::PhantomData, path::PathBuf, sync::Arc, time::Duration};
 
 use concurrency_manager::ConcurrencyManager;
 use engine_traits::KvEngine;
@@ -15,7 +12,6 @@ use kvproto::{
 };
 use online_config::ConfigChange;
 use pd_client::PdClient;
-
 use raftstore::{
     coprocessor::{CmdBatch, ObserveHandle, RegionInfoProvider},
     router::RaftStoreRouter,
@@ -36,7 +32,6 @@ use tokio::{
 use tokio_stream::StreamExt;
 use txn_types::TimeStamp;
 
-
 use super::metrics::HANDLE_EVENT_DURATION_HISTOGRAM;
 use crate::{
     async_task_manager::RegionSubscriptionManager,
@@ -710,7 +705,7 @@ where
                 Error::from(err).report("failed to update service safe point!");
                 // don't give up?
             }
-            if let Err(err) = meta_cli.step_task(&task, rts).await {
+            if let Err(err) = meta_cli.set_local_task_checkpoint(&task, rts).await {
                 err.report(format!("on flushing task {}", task));
                 // we can advance the progress at next time.
                 // return early so we won't be mislead by the metrics.

components/backup-stream/src/metadata/client.rs

@@ -279,7 +279,7 @@ impl<Store: MetaStore> MetadataClient<Store> {
     }
 
     /// forward the progress of some task.
-    pub async fn step_task(&self, task_name: &str, ts: u64) -> Result<()> {
+    pub async fn set_local_task_checkpoint(&self, task_name: &str, ts: u64) -> Result<()> {
         let now = Instant::now();
         defer! {
             super::metrics::METADATA_OPERATION_LATENCY.with_label_values(&["task_step"]).observe(now.saturating_elapsed().as_secs_f64())

components/backup-stream/src/metadata/test.rs

@@ -123,10 +123,10 @@ async fn test_progress() -> Result<()> {
     cli.insert_task_with_range(&task, &[]).await?;
     let progress = cli.progress_of_task(&task.info.name).await?;
     assert_eq!(progress, task.info.start_ts);
-    cli.step_task(&task.info.name, 42).await?;
+    cli.set_local_task_checkpoint(&task.info.name, 42).await?;
     let progress = cli.progress_of_task(&task.info.name).await?;
     assert_eq!(progress, 42);
-    cli.step_task(&task.info.name, 43).await?;
+    cli.set_local_task_checkpoint(&task.info.name, 43).await?;
     let progress = cli.progress_of_task(&task.info.name).await?;
     assert_eq!(progress, 43);
     let other_store = MetadataClient::new(cli.meta_store.clone(), 43);