Repository: tikv/tikv
File: components/resolved_ts/src/scanner.rs (Rust)

Code before:
    }

    async fn get_snapshot(
        task: &mut ScanTask,
        cdc_handle: T,
    ) -> Result<RegionSnapshot<E::Snapshot>> {
        let mut last_err = None;
        for retry_times in 0..=GET_SNAPSHOT_RETRY_TIME {
            if retry_times != 0 {
                if let Err(e) = GLOBAL_TIMER_HANDLE
                    .delay(
                        std::time::Instant::now()
                            + GET_SNAPSHOT_RETRY_BACKOFF_STEP
                                .mul_f64(10_f64.powi(retry_times as i32 - 1)),
                    )
                    .compat()
                    .await
                {
                    error!("failed to backoff"; "err" => ?e);
                }
                if task.is_cancelled() {
                    return Err(box_err!("scan task cancelled"));
                }
            }
            let (cb, fut) = tikv_util::future::paired_future_callback();
            let change_cmd = ChangeObserver::from_rts(task.region.id, task.handle.clone());
            cdc_handle
                .capture_change(
                    task.region.id,
                    task.region.get_region_epoch().clone(),
                    change_cmd,
                    Callback::read(Box::new(cb)),
                )
                .map_err(|e| Error::Other(box_err!("{:?}", e)))?;
            let mut resp = box_try!(fut.await);
            if resp.response.get_header().has_error() {
                let err = resp.response.take_header().take_error();
                // These two errors can't handled by retrying since the epoch and observe id is
                // unchanged
                if err.has_epoch_not_match() || err.get_message().contains("stale observe id") {
                    return Err(box_err!("get snapshot failed: {:?}", err));
                }
                last_err = Some(err)
            } else {
                return Ok(resp.snapshot.unwrap());
            }
        }
        Err(box_err!(
            "backoff timeout after {} try, last error: {:?}",
            GET_SNAPSHOT_RETRY_TIME,
            last_err.unwrap()

Review comment (suggestion):
 It seems all canceled check are after `.await` point, perhaps we can trivially wrap the whole future with [`Abortable`](https://docs.rs/futures/latest/futures/future/struct.Abortable.html).

Code after:
        };
        self.workers.spawn(fut);
    }

    async fn get_snapshot(
        task: &mut ScanTask,
        cdc_handle: T,
    ) -> Result<RegionSnapshot<E::Snapshot>> {
        let mut last_err = None;
        for retry_times in 0..=GET_SNAPSHOT_RETRY_TIME {
            if retry_times != 0 {
                let mut backoff = GLOBAL_TIMER_HANDLE
                    .delay(
                        std::time::Instant::now()
                            + GET_SNAPSHOT_RETRY_BACKOFF_STEP
                                .mul_f64(10_f64.powi(retry_times as i32 - 1)),
                    )
                    .compat()
                    .fuse();
                futures::select! {
                    res = backoff => if let Err(e) = res {
                        error!("failed to backoff"; "err" => ?e);
                    },
                    _ = &mut task.cancelled => {}
                }
                if task.is_cancelled() {
                    return Err(box_err!("scan task cancelled"));
                }
            }
            let (cb, fut) = tikv_util::future::paired_future_callback();
            let change_cmd = ChangeObserver::from_rts(task.region.id, task.handle.clone());
            cdc_handle
                .capture_change(
                    task.region.id,
                    task.region.get_region_epoch().clone(),
                    change_cmd,
                    Callback::read(Box::new(cb)),
                )
                .map_err(|e| Error::Other(box_err!("{:?}", e)))?;
            let mut resp = box_try!(fut.await);
            if resp.response.get_header().has_error() {
                let err = resp.response.take_header().take_error();
                // These two errors can't handled by retrying since the epoch and observe id is
                // unchanged
                if err.has_epoch_not_match() || err.get_message().contains("stale observe id") {
                    return Err(box_err!("get snapshot failed: {:?}", err));
                }
                last_err = Some(err)
            } else {
                return Ok(resp.snapshot.unwrap());
            }