Как отменить Tokio task и что происходит, когда future прерывается в середине выполнения?

Отмена Tokio task и поведение future при прерывании

В Tokio отмена задачи происходит через вызов JoinHandle::abort(). После этого runtime при следующей возможности дропает future задачи. Аналогично future отменяется, когда её дропают из ветки tokio::select!, которая не была выбрана, или когда JoinHandle выбрасывается без .await.

Механизм abort()

use tokio::time::{sleep, Duration};

#[tokio::main]
async fn main() {
    let handle = tokio::spawn(async {
        println!("task started");
        sleep(Duration::from_secs(10)).await;   // точка прерывания
        println!("task finished");               // это не выполнится
    });

    sleep(Duration::from_millis(50)).await;
    handle.abort();

    match handle.await {
        Ok(_) => println!("completed normally"),
        Err(e) if e.is_cancelled() => println!("task was cancelled"),
        Err(e) => println!("task panicked: {e}"),
    }
}

abort() лишь помечает задачу на отмену. Реальная отмена случается в следующей точке .await внутри задачи — именно там Tokio может вернуть Poll::Pending и затем дропнуть future. Код после этой точки await никогда не выполнится.

Что происходит при дропе future

Rust гарантирует вызов деструкторов (Drop) для всех значений, находившихся в стеке future в момент прерывания. Однако любая логическая работа, которая ещё не завершилась, теряется. Пример опасного сценария:

async fn transfer_funds(db: &Db, from: i64, to: i64, amount: u64) {
    db.debit(from, amount).await;   // <-- если задача отменена здесь...
    db.credit(to, amount).await;    // ...этот вызов никогда не произойдёт
}

Деньги спишутся, но не зачислятся. Решение — обернуть операцию в транзакцию БД или использовать CancellationToken с явным завершением.

CancellationToken — graceful shutdown

use tokio_util::sync::CancellationToken;
use tokio::time::{sleep, Duration};

#[tokio::main]
async fn main() {
    let token = CancellationToken::new();
    let child = token.clone();

    let handle = tokio::spawn(async move {
        tokio::select! {
            _ = child.cancelled() => {
                // корректная очистка
                println!("graceful shutdown");
            }
            _ = do_work() => {
                println!("work done");
            }
        }
    });

    sleep(Duration::from_millis(100)).await;
    token.cancel();         // сигнализируем задаче
    handle.await.unwrap();
}

async fn do_work() {
    sleep(Duration::from_secs(60)).await;
}

abort_handle()

Начиная с Tokio 1.17 можно получить AbortHandle отдельно от JoinHandle, что удобно, когда нужно отменить задачу, не дожидаясь её завершения:

let (handle, abort_handle) = {
    let jh = tokio::spawn(long_task());
    let ah = jh.abort_handle();
    (jh, ah)
};
// позже:
abort_handle.abort();
let _ = handle.await; // JoinError::is_cancelled() == true

Подводные камни

• abort() не является синхронным — задача продолжает работать до следующей точки .await. Если задача зависла в CPU-bound цикле без await, она не будет отменена.
• Дроп JoinHandle без .await отсоединяет задачу (detach), а не отменяет её — задача продолжает работать в фоне.
• Деструкторы вызываются, но асинхронный код в Drop не выполняется — нельзя написать async fn drop. Финальная очистка требует явного паттерна (CancellationToken + select).
• После abort() вызов handle.await обязателен для гарантии завершения задачи до выхода из runtime. Иначе runtime просто уничтожит задачу при shutdown без гарантии очистки.
• Паника внутри задачи и отмена возвращают разные варианты JoinError: is_panic() vs is_cancelled() — различайте их явно.
• Если задача захватывает MutexGuard из std через точку await — это deadlock; при отмене guard освобождается через Drop, но это произойдёт в асинхронном контексте без гарантий порядка.

• Отвечать определением без production-сценария.
• Не называть runtime boundary, security boundary или failure mode.
• Игнорировать версию API, observability и тестовую проверку.

• Объясняет механизм своими словами и без выдуманных API.
• Называет реальные риски, диагностику и критерий корректности.
• Связывает ответ с текущей документацией и миграционными ограничениями.