Repository: tikv/tikv (language: rust)

Issue:
log backup status is normal and lag more and more after injection pd leader io delay 500ms last for 10mins and then recover
## Bug Report

<!-- Thanks for your bug report! Don't worry if you can't fill out all the sections. -->

### What version of TiKV are you using?
<!-- You can run `tikv-server --version` -->
master
### What operating system and CPU are you using?
<!-- If you're using Linux, you can run `cat /proc/cpuinfo` -->

### Steps to reproduce
<!-- If possible, provide a recipe for reproducing the error. A complete runnable program is good. -->

### What did you expect?
see https://github.com/pingcap/tidb/issues/58574
### What did happened?
The task is removed from `range_router` before it handle the fatal error. Therefore, the task won't be selected again if it failed to handle the fatal error and retry to schedule the fatal error.


Discussion / hints:

Fix patch:
diff --git a/components/backup-stream/src/endpoint.rs b/components/backup-stream/src/endpoint.rs
index 5619069bebc..18ba29e2131 100644
--- a/components/backup-stream/src/endpoint.rs
+++ b/components/backup-stream/src/endpoint.rs
@@ -32,7 +32,7 @@ use tikv::config::{BackupStreamConfig, ResolvedTsConfig};
 use tikv_util::{
     box_err,
     config::ReadableDuration,
-    debug, defer, info,
+    debug, defer, error, info,
     memory::MemoryQuota,
     sys::thread::ThreadBuildWrapper,
     time::{Instant, Limiter},
@@ -233,8 +233,10 @@ where
         self.meta_client.clone()
     }
 
-    fn on_fatal_error_of_task(&self, task: &str, err: &Error) -> future![()] {
+    fn on_fatal_error_of_task(&self, task: &str, err: &Error) -> future![bool] {
         metrics::update_task_status(TaskStatus::Error, task);
+        error!(?err; "Task encounters fatal error, will pause this task."; "task" => %task,);
+
         let meta_cli = self.get_meta_client();
         let pdc = self.pd_client.clone();
         let store_id = self.store_id;
@@ -246,6 +248,12 @@ where
         let task = task.to_owned();
         async move {
             let err_fut = async {
+                #[cfg(feature = "failpoints")]
+                fail::fail_point!("log-backup-upload-error", |v| {
+                    info!("injected error."; "value" => ?v);
+                    Result::Err(Error::Other(box_err!("injected error: {:?}", v)))
+                });
+
                 let safepoint = meta_cli.global_progress_of_task(&task).await?;
                 pdc.update_service_safe_point(
                     safepoint_name,
@@ -262,7 +270,10 @@ where
                 meta_cli.report_last_error(&task, last_error).await?;
                 Result::Ok(())
             };
-            if let Err(err_report) = err_fut.await {
+            let res = err_fut.await;
+            let paused = res.is_ok();
+
+            if let Err(err_report) = res {
                 err_report.report(format_args!("failed to upload error {}", err_report));
                 let name = task.to_owned();
                 // Let's retry reporting after 5s.
@@ -277,6 +288,7 @@ where
                     );
                 });
             }
+            paused
         }
     }
 
@@ -287,9 +299,12 @@ where
             .block_on(self.range_router.select_task(select.reference()));
         warn!("fatal error reporting"; "selector" => ?select, "selected" => ?tasks, "err" => %err);
         for task in tasks {
-            // Let's pause the task first.
-            self.unload_task(&task);
-            self.pool.block_on(self.on_fatal_error_of_task(&task, &err));
+            if self.pool.block_on(self.on_fatal_error_of_task(&task, &err)) {
+                // It is necessary to notify all other stores before pausing the task locally.
+                // Or once uploading error failed, we cannot upload it because the task cannot
+                // be found.
+                self.unload_task(&task);
+            }
         }
     }
 


Test patch:
diff --git a/components/backup-stream/tests/failpoints/mod.rs b/components/backup-stream/tests/failpoints/mod.rs
index 3d69e5ace7d..ec7416ca313 100644
--- a/components/backup-stream/tests/failpoints/mod.rs
+++ b/components/backup-stream/tests/failpoints/mod.rs
@@ -18,16 +18,19 @@ mod all {
     };
 
     use backup_stream::{
+        errors::Error,
         metadata::{
             keys::MetaKey,
             store::{Keys, MetaStore},
         },
+        router::TaskSelector,
         GetCheckpointResult, RegionCheckpointOperation, RegionSet, Task,
     };
     use futures::executor::block_on;
     use tikv_util::{
+        box_err,
         config::{ReadableDuration, ReadableSize},
-        defer,
+        defer, HandyRwLock,
     };
     use txn_types::Key;
 
@@ -409,4 +412,54 @@ mod all {
         // Make sure our suite doesn't panic.
         suite.sync();
     }
+
+    #[test]
+    fn fatal_error() {
+        let mut suite = SuiteBuilder::new_named("fatal_error").nodes(3).build();
+        suite.must_register_task(1, "test_fatal_error");
+        suite.sync();
+        run_async_test(suite.write_records(0, 1, 1));
+        suite.force_flush_files("test_fatal_error");
+        suite.wait_for_flush();
+        run_async_test(suite.advance_global_checkpoint("test_fatal_error")).unwrap();
+        fail::cfg("log-backup-upload-error", "1*return(not my fault)->off").unwrap();
+        let (victim, endpoint) = suite.endpoints.iter().next().unwrap();
+        endpoint
+            .scheduler()
+            .schedule(Task::FatalError(
+                TaskSelector::ByName("test_fatal_error".to_owned()),
+                Box::new(Error::Other(box_err!("everything is alright"))),
+            ))
+            .unwrap();
+        suite.sync();
+        fail::remove("log-backup-upload-error");
+        // Wait retry...
+        std::thread::sleep(Duration::from_secs(6));
+        suite.sync();
+        let err = run_async_test(
+            suite
+                .get_meta_cli()
+                .get_last_error_of("test_fatal_error", *victim),
+        )
+        .unwrap()
+        .unwrap();
+        assert_eq!(err.error_code, error_code::backup_stream::OTHER.code);
+        assert!(err.error_message.contains("everything is alright"));
+        assert_eq!(err.store_id, *victim);
+        let paused =
+            run_async_test(suite.get_meta_cli().check_task_paused("test_fatal_error")).unwrap();
+        assert!(paused);
+        let safepoints = suite.cluster.pd_client.gc_safepoints.rl();
+        let checkpoint = suite.global_checkpoint();
+
+        assert!(
+            safepoints.iter().any(|sp| {
+                sp.serivce.contains(&format!("{}", victim))
+                    && sp.ttl >= Duration::from_secs(60 * 60 * 24)
+                    && sp.safepoint.into_inner() == checkpoint - 1
+            }),
+            "{:?}",
+            safepoints
+        );
+    }
 }
diff --git a/components/backup-stream/tests/integration/mod.rs b/components/backup-stream/tests/integration/mod.rs
index 6724b31a66a..bee7d4e91f3 100644
--- a/components/backup-stream/tests/integration/mod.rs
+++ b/components/backup-stream/tests/integration/mod.rs
@@ -10,14 +10,11 @@ pub use suite::*;
 mod all {
     use std::time::{Duration, Instant};
 
-    use backup_stream::{
-        errors::Error, router::TaskSelector, GetCheckpointResult, RegionCheckpointOperation,
-        RegionSet, Task,
-    };
+    use backup_stream::{GetCheckpointResult, RegionCheckpointOperation, RegionSet, Task};
     use futures::{Stream, StreamExt};
     use pd_client::PdClient;
     use test_raftstore::IsolationFilterFactory;
-    use tikv_util::{box_err, defer, info, HandyRwLock};
+    use tikv_util::defer;
     use tokio::time::timeout;
     use txn_types::{Key, TimeStamp};
 
@@ -141,54 +138,6 @@ mod all {
         suite.cluster.shutdown();
     }
 
-    #[test]
-    fn fatal_error() {
-        let mut suite = super::SuiteBuilder::new_named("fatal_error")
-            .nodes(3)
-            .build();
-        suite.must_register_task(1, "test_fatal_error");
-        suite.sync();
-        run_async_test(suite.write_records(0, 1, 1));
-        suite.force_flush_files("test_fatal_error");
-        suite.wait_for_flush();
-        run_async_test(suite.advance_global_checkpoint("test_fatal_error")).unwrap();
-        let (victim, endpoint) = suite.endpoints.iter().next().unwrap();
-        endpoint
-            .scheduler()
-            .schedule(Task::FatalError(
-                TaskSelector::ByName("test_fatal_error".to_owned()),
-                Box::new(Error::Other(box_err!("everything is alright"))),
-            ))
-            .unwrap();
-        suite.sync();
-        let err = run_async_test(
-            suite
-                .get_meta_cli()
-                .get_last_error_of("test_fatal_error", *victim),
-        )
-        .unwrap()
-        .unwrap();
-        info!("err"; "err" => ?err);
-        assert_eq!(err.error_code, error_code::backup_stream::OTHER.code);
-        assert!(err.error_message.contains("everything is alright"));
-        assert_eq!(err.store_id, *victim);
-        let paused =
-            run_async_test(suite.get_meta_cli().check_task_paused("test_fatal_error")).unwrap();
-        assert!(paused);
-        let safepoints = suite.cluster.pd_client.gc_safepoints.rl();
-        let checkpoint = suite.global_checkpoint();
-
-        assert!(
-            safepoints.iter().any(|sp| {
-                sp.serivce.contains(&format!("{}", victim))
-                    && sp.ttl >= Duration::from_secs(60 * 60 * 24)
-                    && sp.safepoint.into_inner() == checkpoint - 1
-            }),
-            "{:?}",
-            safepoints
-        );
-    }
-
     #[test]
     fn region_checkpoint_info() {
         let mut suite = super::SuiteBuilder::new_named("checkpoint_info")