Repository: tikv/tikv
File: components/backup-stream/src/endpoint.rs (Rust)

Code before:
            pd_client,
            subs,
            concurrency_manager,
            region_operator,
            failover_time: None,
            config,
            checkpoint_mgr,
            abort_last_storage_save: None,
        };
        ep.pool.spawn(root!(ep.min_ts_worker()));
        ep
    }
}

impl<S, R, E, PDC> Endpoint<S, R, E, PDC>
where
    S: MetaStore + 'static,
    R: RegionInfoProvider + Clone + 'static,
    E: KvEngine,
    PDC: PdClient + 'static,
{
    fn get_meta_client(&self) -> MetadataClient<S> {
        self.meta_client.clone()
    }

    fn on_fatal_error_of_task(&self, task: &str, err: &Error) -> future![()] {
        metrics::update_task_status(TaskStatus::Error, task);
        let meta_cli = self.get_meta_client();
        let pdc = self.pd_client.clone();
        let store_id = self.store_id;
        let sched = self.scheduler.clone();
        let safepoint_name = self.pause_guard_id_for_task(task);
        let safepoint_ttl = self.pause_guard_duration();
        let code = err.error_code().code.to_owned();
        let msg = err.to_string();
        let t = task.to_owned();
        let f = async move {
            let err_fut = async {
                let safepoint = meta_cli.global_progress_of_task(&t).await?;
                pdc.update_service_safe_point(
                    safepoint_name,
                    TimeStamp::new(safepoint.saturating_sub(1)),
                    safepoint_ttl,
                )
                .await
                .or_else(|err| match err {
                    pd_client::Error::UnsafeServiceGcSafePoint { .. } => {
                        warn!("gc safe point exceeds the task checkpoint. skipping uploading"; "err" => %err);
                        Ok(())
                    }
                    _ => Err(err),

Review comment (suggestion):
since we are modifying this part, can we add a fatal/error log in this method, I noticed that it only uploads the status but in the log we can't see the error. It could be easier when looking at TiKV logs searching for error keyword. 

Code after:
            pd_client,
            subs,
            concurrency_manager,
            region_operator,
            failover_time: None,
            config,
            checkpoint_mgr,
            abort_last_storage_save: None,
        };
        ep.pool.spawn(root!(ep.min_ts_worker()));
        ep
    }
}

impl<S, R, E, PDC> Endpoint<S, R, E, PDC>
where
    S: MetaStore + 'static,
    R: RegionInfoProvider + Clone + 'static,
    E: KvEngine,
    PDC: PdClient + 'static,
{
    fn get_meta_client(&self) -> MetadataClient<S> {
        self.meta_client.clone()
    }

    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;
        let sched = self.scheduler.clone();
        let safepoint_name = self.pause_guard_id_for_task(task);
        let safepoint_ttl = self.pause_guard_duration();
        let code = err.error_code().code.to_owned();
        let msg = err.to_string();
        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,
                    TimeStamp::new(safepoint.saturating_sub(1)),
                    safepoint_ttl,