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

Code before:
                } else {
                    let sched = self.scheduler.clone();
                    self.pool.spawn(root!(async move {
                        tokio::time::sleep(Duration::from_millis(500)).await;
                        sched.schedule(Task::Sync(cb, cond)).unwrap();
                    }));
                }
            }
            Task::MarkFailover(t) => self.failover_time = Some(t),
            Task::ExecFlush(task, min_ts) => self.on_exec_flush(task, min_ts),
            Task::RegionCheckpointsOp(s) => self.handle_region_checkpoints_op(s),
            Task::UpdateGlobalCheckpoint(task) => self.on_update_global_checkpoint(task),
            Task::Flushed(result) => self.on_flushed(result),
        }
    }

    fn on_flushed(&mut self, result: FlushResult) {
        if let Some(sender) = self.flush_done_subscribers.remove(&result.task) {
            // Send the message after the subscription manager have tried to sent this flush
            // result to subscribers.
            self.checkpoint_mgr.sync_with_subs_mgr(move |_| {
            if let Err(err) = sender.try_send(result) {
                let err_msg = err.to_string();
                info!("failed to send flush result, waiter is gone or channel blocked"; "err" => %err_msg, "result" => ?err.into_inner());
            }
        })
        }
    }

    fn min_ts_worker(&self) -> future![()] {
        let sched = self.scheduler.clone();
        let interval = self.config.min_ts_interval.0;
        async move {
            loop {
                tokio::time::sleep(interval).await;
                try_send!(
                    sched,
                    Task::RegionCheckpointsOp(RegionCheckpointOperation::PrepareMinTsForResolve)
                );
            }
        }
    }

    pub fn handle_region_checkpoints_op(&mut self, op: RegionCheckpointOperation) {
        match op {
            RegionCheckpointOperation::Resolved {
                checkpoints,
                start_time,
            } => {
                self.checkpoint_mgr.resolve_regions(checkpoints);
                metrics::MIN_TS_RESOLVE_DURATION.observe(start_time.saturating_elapsed_secs());

Review comment (bug):
```suggestion
            self.checkpoint_mgr.sync_with_subs_mgr(move |_| {
                if let Err(err) = sender.try_send(result) {
                    let err_msg = err.to_string();
                    info!("failed to send flush result, waiter is gone or channel blocked"; "err" => %err_msg, "result" => ?err.into_inner());
                }
            })
```


Code after:
                } else {
                    let sched = self.scheduler.clone();
                    self.pool.spawn(root!(async move {
                        tokio::time::sleep(Duration::from_millis(500)).await;
                        sched.schedule(Task::Sync(cb, cond)).unwrap();
                    }));
                }
            }
            Task::MarkFailover(t) => self.failover_time = Some(t),
            Task::ExecFlush(task, min_ts) => self.on_exec_flush(task, min_ts),
            Task::RegionCheckpointsOp(s) => self.handle_region_checkpoints_op(s),
            Task::UpdateGlobalCheckpoint(task) => self.on_update_global_checkpoint(task),
            Task::Flushed(result) => self.on_flushed(result),
        }
    }

    fn on_flushed(&mut self, result: FlushResult) {
        if let Some(sender) = self.flush_done_subscribers.remove(&result.task) {
            // Send the message after the subscription manager have tried to sent this flush
            // result to subscribers.
            self.checkpoint_mgr.sync_with_subs_mgr(move |_| {
                if let Err(err) = sender.try_send(result) {
                    let err_msg = err.to_string();
                    info!("failed to send flush result, waiter is gone or channel blocked"; "err" => %err_msg, "result" => ?err.into_inner());
                }
            })
        }
    }

    fn min_ts_worker(&self) -> future![()] {
        let sched = self.scheduler.clone();
        let interval = self.config.min_ts_interval.0;
        async move {
            loop {
                tokio::time::sleep(interval).await;
                try_send!(
                    sched,
                    Task::RegionCheckpointsOp(RegionCheckpointOperation::PrepareMinTsForResolve)
                );
            }
        }
    }

    pub fn handle_region_checkpoints_op(&mut self, op: RegionCheckpointOperation) {
        match op {
            RegionCheckpointOperation::Resolved {
                checkpoints,
                start_time,
            } => {
                self.checkpoint_mgr.resolve_regions(checkpoints);
                metrics::MIN_TS_RESOLVE_DURATION.observe(start_time.saturating_elapsed_secs());