Cancelling is dropping what stops in Tokio, and what leaks

Seven tested behaviours of timeout, select!, JoinSet, JoinHandle, spawn_blocking and CancellationToken, and the one rule that keeps a job queue from losing work.

The problem

A Rust service pulls jobs from an in-memory queue and runs each one. It has three rules. Slow work must give up after a deadline. Shutdown must stop everything cleanly. A job must never disappear without being run or put back.

The team reaches for the obvious tools. They wrap slow calls in tokio::time::timeout. They write a worker loop with tokio::select! so it can wake on either a new job or a periodic health tick. They keep JoinHandles for background tasks and drop them when a request ends.

Then the queue starts leaking. A dashboard shows jobs taken from the queue that never finish and never come back. Nothing panics and no error is logged. A background task that was supposed to end with its request keeps writing to the database ten seconds later. On shutdown, one task refuses to stop at all.

None of these are Tokio bugs. Each one is a precise rule about what "cancel" means in async Rust, and each rule is easy to test once you know it. This lesson runs seven small tests against Tokio 1.54.1 and quotes the real output, so you can see each rule happen.

The idea in plain words

A future is a value that represents work not yet finished. Calling an async fn builds a future; nothing runs until something polls it. An **.await point** is where a future may pause and hand control back to the runtime.

In async Rust, cancellation means dropping a future before it finishes. There is no exception and no signal. The runtime simply stops polling it and runs its destructors. Because a future only pauses at .await, it can only be stopped there. Any state it changed before that point stays changed.

A task is different. tokio::spawn hands a future to the runtime, which polls it independently and returns a **JoinHandle**, a handle you can await for the result. Dropping the handle does not stop the task. You stop a task with abort, or by dropping a container that aborts on drop.

Cancel safety is the property that dropping a future at any .await loses nothing. A cancel-safe operation either completed or left the world as it found it.

The lab tracks drops with a small guard: a struct that bumps a counter in its Drop unless the work called disarm() first. A count of 1 means one future was dropped before it finished.

timeout drops the inner future

timeout(duration, fut) races a future against a timer. The tests use a paused test clock (#[tokio::test(start_paused = true)], which needs the test-util feature): time only moves when every task is idle, so a ten-second sleep finishes instantly and in a fixed order.

let drops = DropCounter::default();
let g = drops.guard();
let slow = async move {
    sleep(Duration::from_secs(10)).await;
    g.disarm();
    "done"
};
let r = timeout(Duration::from_secs(2), slow).await;

t1 result: Err(Elapsed(()))
t1 inner futures dropped before finishing: 1

The inner future was waiting at its sleep when the two-second timer fired. timeout returned Err(Elapsed(())) and dropped the future, so the guard never reached disarm() and the counter went to 1. The Tokio source says this directly: "Cancelling a timeout is done by dropping the future. No additional cleanup or other work is required." The docs add one caveat worth remembering: "the future is polled before the timeout is checked", so a future that never yields can run past its deadline. A timeout only bites at an .await.

A future that changes state before it awaits

select! polls several futures at once and runs the branch of whichever finishes first. Every other branch is dropped. That is cancellation, and it happens on every loop iteration that a branch loses.

async fn take_then_wait(q: Arc<Mutex<Vec<u32>>>) -> Option<u32> {
    let item = q.lock().unwrap().pop(); // state change happens here
    sleep(Duration::from_millis(50)).await; // cancellation point after the change
    item
}

for _ in 0..3 {
    tokio::select! {
        v = take_then_wait(q.clone()) => { if let Some(v) = v { got.push(v) } }
        _ = sleep(Duration::from_millis(10)) => {}
    }
}

t2 delivered: [], left in queue: []

The queue started as [1, 2, 3]. Each iteration, take_then_wait popped an item and then paused for 50 ms. The 10 ms timer won every race, so select! dropped the job future while it held the popped item. Three loops later the queue is empty and nothing was delivered. This is the leaking-queue symptom from the problem section, reproduced in about a dozen lines. The function is not cancel safe because it mutates shared state and then awaits.

mpsc recv is cancel safe

Tokio's bounded channel, tokio::sync::mpsc, is a queue with a sender side and a receiver side. Its recv is the same shape of race, but it behaves differently.

let (tx, mut rx) = mpsc::channel::<u32>(8);
tokio::spawn(async move {
    for i in 1..=3 {
        sleep(Duration::from_millis(50)).await;
        tx.send(i).await.unwrap();
    }
});
loop {
    tokio::select! {
        v = rx.recv() => match v { Some(v) => got.push(v), None => break },
        _ = sleep(Duration::from_millis(10)) => { ticks += 1; }
    }
}

t3 delivered: [1, 2, 3], timer ticks won: 15

The timer won the race fifteen times, so recv was dropped fifteen times. Every message still arrived, in order. The guarantee is written in the method's docs: "This method is cancel safe. If recv is used as a branch in tokio::select! and another branch completes first, it is guaranteed that no messages were received on this channel." The channel only removes a message in the same step that hands it to you, so there is no window where the item is out of the queue but not yet yours.

The select! docs keep a list. Cancel safe: mpsc::Receiver::recv, broadcast::Receiver::recv, watch::Receiver::changed, TcpListener::accept, AsyncReadExt::read, AsyncWriteExt::write, and StreamExt::next on any stream. Not cancel safe, with possible data loss: read_exact, read_to_end, read_to_string, write_all. Also not cancel safe, because "cancellation makes you lose your place in the queue": Mutex::lock, RwLock::read, RwLock::write, Semaphore::acquire, and Notify::notified.

JoinSet: completion order, and drop aborts the rest

A **JoinSet** is a collection of spawned tasks that you drain with join_next(), which returns whichever task finishes next.

for (id, ms) in [(1u32, 300u64), (2, 100), (3, 200), (4, 5_000), (5, 6_000)] {
    let g = drops.guard();
    set.spawn(async move {
        sleep(Duration::from_millis(ms)).await;
        g.disarm();
        id
    });
}
for _ in 0..3 {
    order.push(set.join_next().await.unwrap().unwrap());
}
drop(set);

t4 first three by completion: [2, 3, 1], still in set: 2
t4 tasks dropped unfinished after drop(set): 2

The tasks came back in the order they finished (100 ms, 200 ms, 300 ms), not the order they were spawned. Two slow tasks were still inside the set. Dropping the set aborted both, and their guards counted two unfinished drops. The docs say: "When the JoinSet is dropped, all tasks in the JoinSet are immediately aborted." That makes a JoinSet a good owner for request-scoped work: when the request handler returns, the set goes out of scope and its children stop at their next .await.

Dropping a JoinHandle detaches; abort cancels

A bare JoinHandle works the other way.

let h = tokio::spawn(async move {
    sleep(Duration::from_millis(100)).await;
    *r2.lock().unwrap() = true;
});
drop(h);
sleep(Duration::from_millis(200)).await;

let h2 = tokio::spawn(async { sleep(Duration::from_secs(60)).await; 7 });
h2.abort();
let e = h2.await.unwrap_err();

t5 detached task still ran: true
t5 aborted: is_cancelled=true msg=task 8 was cancelled

The first task kept running after its handle was gone and set the flag. The docs call this detaching: "A JoinHandle detaches the associated task when it is dropped, which means that there is no longer any handle to the task, and no way to join on it." That explains the background write that outlived its request. The second task was stopped with abort(). Awaiting its handle returned a JoinError, and is_cancelled() reported true. Check that method when you collect results, so a deliberate abort is not logged as a crash.

spawn_blocking ignores abort once it runs

spawn_blocking runs a plain, non-async closure on a separate thread pool meant for blocking work. This test uses a real multi-thread runtime, because the paused clock does not apply to OS threads.

let h = tokio::task::spawn_blocking(move || {
    tx.send(()).unwrap(); // tell the test we started
    std::thread::sleep(Duration::from_millis(200));
    42
});
rx.recv().unwrap();
h.abort();
let r = h.await;

t6 spawn_blocking after abort: Ok(42)

The closure had started, so abort() did nothing and the handle returned Ok(42). A blocking closure has no .await points, so there is nowhere to stop it. The source states it plainly: "If you call abort on a spawn_blocking task, then this will not have any effect, and the task will continue running normally. The exception is if the task has not started running yet." This is the task that refused to stop on shutdown. If blocking work must be stoppable, it has to check a flag itself between chunks.

CancellationToken for shutdown trees

tokio_util::sync::CancellationToken is a shared flag that tasks can await. child_token() makes a token that is cancelled when its parent is, without the reverse.

let root = CancellationToken::new();
let a = root.child_token();
let b = root.child_token();
a.cancel();
let worker_b = b.clone();
let h = tokio::spawn(async move {
    tokio::select! {
        _ = worker_b.cancelled() => "stopped by token",
        _ = sleep(Duration::from_secs(3600)) => "ran to the end",
    }
});
root.cancel();

t7 after a.cancel(): root=false a=true b=false
t7 after root.cancel(): b=true worker said: stopped by token

Cancelling child a touched nothing else. Cancelling the root reached b, and the worker's select! took the token branch instead of sleeping for an hour. This gives a service one shutdown switch at the top and narrower switches per subsystem. The token asks; it does not force. The worker chooses where it checks, so it can finish the job in hand first.

Making your own future cancel safe

The fix for take_then_wait follows from the rule: nothing that matters may change before the last .await that can be dropped. Three patterns work. Move the state change after the await. Replace the shared Vec with a cancel-safe primitive such as mpsc. Or take the item outside the select!, so the race only decides what to do with an item you already hold.

This version uses the third pattern. It was not run in this lab.

loop {
    let Some(job) = rx.recv().await else { break }; // cancel safe
    tokio::select! {
        r = run(job) => record(r),
        _ = shutdown.cancelled() => { requeue(job_id).await; break }
    }
}

Taking the job is now its own step, outside the race. If shutdown wins, the loop still knows which job it held and can put it back. requeue and job_id are placeholders for your own code. If run itself mutates state before an await, the same audit applies inside it.

What stops, what keeps running

Limits

The paused clock made sleeps instant and the race order fixed; real timers make race order less predictable. Test t6 used a real two-worker runtime. Not tested here: cancel safety of I/O methods such as read_exact, loss of queue position for Mutex::lock, shutdown of a real HTTP server, release builds, and any performance measurement. The fixed loop above was not compiled.

Check yourself

1. A timeout fires while the inner future is in the middle of a CPU loop with no .await. When does it stop? *At the next .await it reaches, or never if it finishes first; the timeout is only checked between polls.*
2. Why did t2 lose all three items while t3 lost none? *take_then_wait popped the item before awaiting, so dropping it discarded the item; recv only removes a message in the same step it returns it.*
3. A request handler spawns a task with tokio::spawn and returns early. Does the task stop? *No. Dropping the JoinHandle detaches it; use a JoinSet or call abort.*
4. You call abort on a running spawn_blocking task during shutdown. What happens? Nothing; it runs to completion. The closure has to check a flag to stop early.

Sources

• tokio::select!, Cancellation safety (1.54.1)
• tokio::time::timeout (1.54.1)
• tokio::task::JoinSet (1.54.1)
• tokio::task::JoinHandle (1.54.1)
• tokio::task::spawn_blocking (1.54.1)
• tokio_util CancellationToken (0.7.20)
• Lab: crate/ (tokio-cancel-lab), cargo test -- --nocapture --test-threads=1, 7 passed, output in crate/lab-test.log.