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 af8f3a99de8..fe8d12c999e 100644
--- a/components/backup-stream/src/endpoint.rs
+++ b/components/backup-stream/src/endpoint.rs
@@ -34,7 +34,7 @@ use tikv::{
 use tikv_util::{
     box_err,
     config::ReadableDuration,
-    debug, defer, info,
+    debug, defer, error, info,
     memory::MemoryQuota,
     sys::thread::ThreadBuildWrapper,
     time::{Instant, Limiter},
@@ -227,8 +227,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;
@@ -240,6 +242,12 @@ where
         let t = task.to_owned();
         let f = 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(&t).await?;
                 pdc.update_service_safe_point(
                     safepoint_name,
@@ -261,9 +269,13 @@ where
                 last_error.set_store_id(store_id);
                 last_error.set_happen_at(TimeStamp::physical_now());
                 meta_cli.report_last_error(&t, 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 = t.to_owned();
                 // Let's retry reporting after 5s.
@@ -278,6 +290,7 @@ where
                     );
                 });
             }
+            paused
         };
         tracing_active_tree::frame!("on_fatal_error_of_task"; f; %err, %task)
     }
@@ -287,9 +300,12 @@ where
         let tasks = 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 2499a679e38..fc0a6fdb9d7 100644
--- a/components/backup-stream/tests/failpoints/mod.rs
+++ b/components/backup-stream/tests/failpoints/mod.rs
@@ -18,10 +18,12 @@ mod all {
     };
 
     use backup_stream::{
+        errors::Error,
         metadata::{
             keys::MetaKey,
             store::{Keys, MetaStore},
         },
+        router::TaskSelector,
         GetCheckpointResult, RegionCheckpointOperation, RegionSet, Task,
     };
     use encryption::{FileConfig, MasterKeyConfig};
@@ -30,8 +32,9 @@ mod all {
     use raftstore::coprocessor::ObserveHandle;
     use tempfile::TempDir;
     use tikv_util::{
+        box_err,
         config::{ReadableDuration, ReadableSize},
-        defer,
+        defer, HandyRwLock,
     };
     use txn_types::Key;
     use walkdir::WalkDir;
@@ -485,4 +488,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.service.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 8b33a46558f..62b07646e37 100644
--- a/components/backup-stream/tests/integration/mod.rs
+++ b/components/backup-stream/tests/integration/mod.rs
@@ -13,15 +13,14 @@ mod all {
     };
 
     use backup_stream::{
-        errors::Error, router::TaskSelector, GetCheckpointResult, RegionCheckpointOperation,
-        RegionSet, Task,
+        router::TaskSelector, GetCheckpointResult, RegionCheckpointOperation, RegionSet, Task,
     };
     use futures::{Stream, StreamExt};
     use kvproto::metapb::RegionEpoch;
     use pd_client::PdClient;
     use test_raftstore::IsolationFilterFactory;
     use tikv::config::BackupStreamConfig;
-    use tikv_util::{box_err, defer, info, HandyRwLock};
+    use tikv_util::defer;
     use tokio::time::timeout;
     use txn_types::{Key, TimeStamp};
     use walkdir::WalkDir;
@@ -200,52 +199,6 @@ mod all {
         suite.cluster.shutdown();
     }
 
-    #[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();
-        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.service.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 = SuiteBuilder::new_named("checkpoint_info").nodes(1).build();