Tokio select! Cancellation Safety and biased
You reach for tokio::select! because it reads like the obvious tool for the job: race a socket read against a shutdown signal, take whichever finishes first, move on. Then production starts losing bytes. A request body arrives in two TCP segments, your shutdown branch wins the race between them, and the half-read buffer evaporates with the dropped future. Worse, the bug only shows up under load, only on certain kernels, and only when the shutdown channel happens to be hot. You read the docs again, find the words "cancellation safety" buried in a sentence, and realise the primitive you thought was a simple match is actually a contract you never signed.
This article rebuilds that contract from the ground up in Rust on top of tokio 1.x, using AsyncReadExt, tokio::sync::mpsc, and tokio::net::TcpStream as the running example. You will scaffold a small binary that reproduces the dropped-read hazard against a real loopback socket, then refactor the unsafe branch into a buffered-read helper whose state survives an arbitrary number of cancellations. From there you will switch to biased select! to make poll ordering deterministic, wire a shutdown-signal plus work-loop pattern around that priority, and finish with a Criterion-style microbenchmark that quantifies the cost of fair versus biased selection under contention. Every step lands as a commit you can check out and run.
It is written for Rust developers who have shipped async code but have never had to defend it against cancellation. By the end you will own a vetted pattern for cancel-safe select!, know which standard tokio futures are safe to drop mid-poll and which are not, and have a benchmark harness you can point at your own hot loops to decide whether biased ordering is worth the fairness trade-off.
Step 1: Wiring a Baseline select! Race Between Two Async Branches
Before we can talk about cancellation safety or biased polling, we need a runnable artifact that puts two futures in the same tokio::select! and lets one of them win. This step builds the smallest possible such artifact: a binary called race-demo that fires two sleep futures with different durations and reports which branch completed first.
We deliberately keep the surface tiny here. There is no networking, no shared state, and no shutdown logic — just one select! macro, one enum, and two tests. That gives later steps a stable reference point: when we start introducing TCP reads, buffered helpers, and biased; ordering, we will be modifying this binary, not a fresh one, so the diffs in subsequent commits stay readable.
Setup
Create the crate skeleton at the root of codebase/. We only need one runtime dependency (tokio with the full feature flag) and one dev-dependency (tokio again, this time with test-util so we can drive virtual time inside #[tokio::test]). The Cargo.toml also declares a single binary target named race-demo pointing at src/main.rs, which is the file that will hold every subsequent step's edits.
[package]
name = "tokio-select-cancellation-safety-biased"
version = "0.1.0"
edition = "2021"
description = "Companion crate for a deep dive on tokio::select! cancellation safety and biased polling"
publish = false
[dependencies]
tokio = { version = "1", features = ["full"] }
[dev-dependencies]
tokio = { version = "1", features = ["full", "test-util"] }
[[bin]]
name = "race-demo"
path = "src/main.rs"
Pulling in full is overkill for this step alone, but the later steps need net, io-util, sync, signal, and macros together — switching to a curated feature list now would mean rewriting the manifest twice. We treat full as the cheap default for a learning crate and revisit feature pruning only if it ever matters for binary size.
Implementation
The whole step lives in src/main.rs. We start with a Winner enum so the race result is a real value, not a stringly-typed log line. Returning an enum means the unit tests can assert against Winner::Fast / Winner::Slow directly instead of parsing stdout.
use std::time::Duration;
use tokio::time::sleep;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Winner {
Fast,
Slow,
}
Next, the race itself. race_branches takes two Duration arguments and hands both sleep futures to tokio::select!. The macro polls every branch, registers wakers, and resolves with the body of whichever branch completes first — the runner-up future is dropped in place. That drop is exactly the behavior we will interrogate in step 2; for now we just need to see it work on harmless sleep futures, which are documented as cancellation-safe.
pub async fn race_branches(fast: Duration, slow: Duration) -> Winner {
tokio::select! {
_ = sleep(fast) => Winner::Fast,
_ = sleep(slow) => Winner::Slow,
}
}
The #[tokio::main] entry point picks deterministic durations (25 ms vs 100 ms) and prints the winning variant. Hardcoded numbers keep the binary reproducible without command-line parsing — when later steps want a configurable demo, we will introduce clap rather than wire up std::env::args by hand.
#[tokio::main]
async fn main() {
let fast = Duration::from_millis(25);
let slow = Duration::from_millis(100);
let winner = race_branches(fast, slow).await;
println!(
"race finished — winner = {winner:?} (fast = {fast:?}, slow = {slow:?})"
);
}
Finally, two tests pin the contract: the shorter duration always wins, regardless of which branch holds it. #[tokio::test(start_paused = true)] freezes the runtime's clock and advances it only when a future is polled, so a 60-second sleep finishes in microseconds of wall time. That keeps the suite fast and removes any flakiness from real timer drift.
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test(start_paused = true)]
async fn fast_branch_wins_when_its_sleep_is_shorter() {
let winner = race_branches(
Duration::from_millis(10),
Duration::from_millis(500),
)
.await;
assert_eq!(winner, Winner::Fast);
}
#[tokio::test(start_paused = true)]
async fn slow_branch_wins_when_fast_is_longer() {
let winner = race_branches(
Duration::from_secs(60),
Duration::from_millis(5),
)
.await;
assert_eq!(winner, Winner::Slow);
}
}
The second test deliberately swaps the durations to prove Winner::Slow is reachable — without it, a buggy implementation that always returned Winner::Fast would still pass the first test.
Verification
Run the test suite and the binary in turn. The tests confirm the race resolves to the correct variant under both orderings; the binary confirms the executable end-to-end path actually compiles and prints the expected line.
cargo test
Finished `test` profile [unoptimized + debuginfo] target(s) in 0.20s
Running unittests src/main.rs (target/debug/deps/race_demo-331142ba86e57cbe)
running 2 tests
test tests::slow_branch_wins_when_fast_is_longer ... ok
test tests::fast_branch_wins_when_its_sleep_is_shorter ... ok
test result: ok. 2 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s
cargo run --quiet
race finished — winner = Fast (fast = 25ms, slow = 100ms)
What we built
We now have a single-file Rust binary that demonstrates the core mechanic of tokio::select!: take N futures, poll them concurrently, return the body of the first one to complete, and drop the rest. That last clause — drop the rest — is the load-bearing detail for every remaining step in this article.
We also locked in a test-time invariant: the race contract is exercised in both directions. Whatever we change in later steps, the fast / slow assertions guarantee we have not accidentally hardcoded the winner or broken the Duration plumbing.
Finally, by adopting start_paused = true from step 1, we sidestep the most common cause of flaky tokio test suites — real clocks racing with CI schedulers. Every future test in this crate will use the same pattern, so contributors do not have to relearn the convention.
The next step takes the same race_branches shape and replaces one of the sleep futures with a partial TCP read. That swap is where the cancellation hazard finally shows up, and we will need this baseline to compare against.
Repository
The state of the code after this step: 0d4ce0a
Step 2: Stranding Partial read_exact Bytes When a select! Timeout Wins
Step 1 left us with a tokio::select! racing two sleep futures. Both are explicitly documented as cancellation-safe, so the macro could drop the losing branch with no observable damage. In this step we deliberately break that property by swapping one arm for tokio::io::AsyncReadExt::read_exact against a real TCP stream, then arrange for the timeout arm to win mid-transfer.
The point is not to ship a useful network client. The point is to make the data loss visible end-to-end, so the rest of the article can reason about a concrete failure mode rather than a hand-wavy "be careful with select!" warning. We will stand up a tiny in-process listener that emits two chunks on a fixed schedule, race a 16-byte read_exact against a short budget, and then drain the socket after the timeout to count exactly how many bytes were stranded.
Setup
No new dependencies are required — the tokio = { features = ["full"] } declaration from step 1 already pulls in tokio::net, tokio::io, and the AsyncReadExt / AsyncWriteExt traits. All edits stay inside the single binary src/main.rs so the diff against step 1 stays small.
We add four top-level items: a ReadOutcome enum capturing the three terminal states of the race, a spawn_chunked_writer helper that boots a one-shot TCP listener emitting a caller-supplied list of chunks, the racing_read_exact function that owns the new select! macro, and a drain_with_budget test-only helper that bounds the post-timeout drain so a stuck socket cannot wedge CI. Two new #[tokio::test] blocks exercise the hazard and the happy path respectively.
use std::net::SocketAddr;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};
These four imports are the entire delta at the top of the file. Everything else — Duration, sleep, Winner, race_branches, the existing tests — survives untouched from step 1.
Implementation
We start with the outcome type. Splitting Completed, Failed, and TimedOut lets each test assert against a precise variant rather than mixing socket errors with deadline expiry. read_exact returns Result<usize, io::Error>, so the Ok / Err projection happens inside the select! arm rather than leaking up to the caller.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ReadOutcome {
Completed,
Failed,
TimedOut,
}
Next, the writer helper. We bind to 127.0.0.1:0 so the kernel picks a free port, hand the resulting SocketAddr back to the caller, and spawn a task that accepts exactly one connection and walks the chunk list. The first chunk goes out immediately; every subsequent chunk waits inter_chunk_delay first. After the last chunk we shutdown() the half-socket so the reader sees a clean EOF instead of hanging on a peer that never closes.
pub async fn spawn_chunked_writer(
chunks: Vec<Vec<u8>>,
inter_chunk_delay: Duration,
) -> std::io::Result<SocketAddr> {
let listener = TcpListener::bind("127.0.0.1:0").await?;
let addr = listener.local_addr()?;
tokio::spawn(async move {
let Ok((mut socket, _)) = listener.accept().await else {
return;
};
for (index, chunk) in chunks.iter().enumerate() {
if index > 0 {
sleep(inter_chunk_delay).await;
}
if socket.write_all(chunk).await.is_err() {
return;
}
let _ = socket.flush().await;
}
let _ = socket.shutdown().await;
});
Ok(addr)
}
The let Ok(...) = ... else { return }; shape keeps the helper at one level of nesting, which is the upper bound the codebase rules allow. Errors from write_all and shutdown are deliberately swallowed: the spawned task has no caller to bubble them up to, and the reader side is what every test actually asserts against. If the listener dies prematurely, the reader's next read returns Ok(0) and the test fails with a clear assertion mismatch rather than a panic in a detached task.
Now the racy reader. The whole function is two lines of body — read_exact on one arm, sleep(budget) on the other — and the only translation step is folding the Result into the appropriate ReadOutcome variant. That brevity is the point: we want the hazard to be attributable to select! itself, not to anything clever in our wrapper.
pub async fn racing_read_exact(
stream: &mut TcpStream,
buf: &mut [u8],
budget: Duration,
) -> ReadOutcome {
tokio::select! {
result = stream.read_exact(buf) => match result {
Ok(_) => ReadOutcome::Completed,
Err(_) => ReadOutcome::Failed,
},
_ = sleep(budget) => ReadOutcome::TimedOut,
}
}
When the sleep arm wins, the read_exact future is dropped in place. read_exact is built on top of repeated poll_read calls into the caller's buf, accumulating a private cursor of how many bytes it has written so far. That cursor lives inside the future state machine — when the future is dropped, the cursor goes with it. The bytes already deposited into buf physically remain in the caller's buffer, but the caller has no way to know how many of those leading bytes are valid because the cursor is gone. That asymmetry is the cancellation hazard in its raw form.
The hazard test wires up a writer that emits two 8-byte chunks with a 300 ms gap, then asks racing_read_exact to pull all 16 bytes inside a 120 ms budget. The first chunk arrives within ~milliseconds and read_exact happily drains it into the front half of buf. The second chunk is still 180 ms away when the sleep budget fires, so select! discards the in-flight read_exact. We then drain whatever is left on the wire — only the 8 bytes of chunk 2 survive, which is the receipt proving that chunk 1's 8 bytes were stranded in the dropped future's user-space buffer.
#[tokio::test]
async fn racing_read_exact_strands_bytes_when_timeout_wins() {
let addr = spawn_chunked_writer(
vec![vec![0xAA; 8], vec![0xBB; 8]],
Duration::from_millis(300),
)
.await
.expect("listener bind failed");
let mut stream = TcpStream::connect(addr)
.await
.expect("client connect failed");
let mut buf = [0u8; 16];
let outcome = racing_read_exact(
&mut stream,
&mut buf,
Duration::from_millis(120),
)
.await;
assert_eq!(outcome, ReadOutcome::TimedOut);
let leftover = drain_with_budget(&mut stream, Duration::from_secs(2)).await;
assert_eq!(
leftover.len(),
8,
"expected exactly 8 leftover bytes (only chunk 2 survives the cancellation), got {}",
leftover.len()
);
assert_eq!(leftover, vec![0xBB; 8]);
}
The two assertions on leftover are what turn this from "looks broken" into a binary signal. If chunk 1's bytes had been left in the kernel — say, if read_exact had been polite enough to back its cursor out before being dropped — we would observe 16 leftover bytes, not 8. Conversely if a future tokio version somehow made read_exact cancellation-safe by surfacing the partial count, the test would still pass for outcome but we would also need to recover those 8 bytes from buf directly, which is exactly the refactor step 3 will perform.
The companion happy-path test (racing_read_exact_completes_when_budget_outlasts_both_chunks) gives the writer a 20 ms inter-chunk gap and a 2 s budget, then asserts the outcome is Completed and buf holds the concatenated payload. Keeping both directions in the suite protects later steps: when we eventually rewrite the reader to hold the in-flight read across the select! boundary, we need a guarantee that the non-pathological case still works, not just that the pathological case is fixed.
#[tokio::test]
async fn racing_read_exact_completes_when_budget_outlasts_both_chunks() {
let addr = spawn_chunked_writer(
vec![vec![0xAA; 8], vec![0xBB; 8]],
Duration::from_millis(20),
)
.await
.expect("listener bind failed");
let mut stream = TcpStream::connect(addr)
.await
.expect("client connect failed");
let mut buf = [0u8; 16];
let outcome = racing_read_exact(
&mut stream,
&mut buf,
Duration::from_secs(2),
)
.await;
assert_eq!(outcome, ReadOutcome::Completed);
assert_eq!(&buf[..8], &[0xAA; 8]);
assert_eq!(&buf[8..], &[0xBB; 8]);
}
The drain_with_budget helper wraps a read-loop in tokio::time::timeout so a misbehaving writer cannot stall the test runner forever. The 2 s ceiling is wildly larger than the 300 ms inter-chunk gap, so under expected conditions the loop exits on Ok(0) long before the budget matters; the ceiling exists purely as a CI safety net.
async fn drain_with_budget(stream: &mut TcpStream, budget: Duration) -> Vec<u8> {
tokio::time::timeout(budget, async {
let mut sink = Vec::new();
let mut tmp = [0u8; 64];
loop {
match stream.read(&mut tmp).await {
Ok(0) => break,
Ok(n) => sink.extend_from_slice(&tmp[..n]),
Err(_) => break,
}
}
sink
})
.await
.unwrap_or_default()
}
Note that these network tests deliberately do not use start_paused = true. Paused-time mode treats real TcpStream I/O as a poll that never advances the virtual clock, so the timer arm never fires and the test hangs forever. Real wall-clock time is the right primitive here precisely because we are exercising the interaction between the OS-level kernel buffer and tokio's timer driver.
Verification
Run the full test suite from the crate root. All four tests — the two sleep-only races from step 1 plus the two new TCP-backed tests — must pass.
cargo test
Finished `test` profile [unoptimized + debuginfo] target(s) in 0.42s
Running unittests src/main.rs (target/debug/deps/race_demo-331142ba86e57cbe)
running 4 tests
test tests::fast_branch_wins_when_its_sleep_is_shorter ... ok
test tests::slow_branch_wins_when_fast_is_longer ... ok
test tests::racing_read_exact_completes_when_budget_outlasts_both_chunks ... ok
test tests::racing_read_exact_strands_bytes_when_timeout_wins ... ok
test result: ok. 4 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.52s
The decisive signal is that racing_read_exact_strands_bytes_when_timeout_wins passes — that is the test that flips to a failure if the cancellation hazard ever disappears (for example, because a hypothetical future tokio version made read_exact magically surface its partial cursor on drop, which today's tokio does not).
What we built
We now have an executable reproduction of the tokio::select! cancellation hazard. The reproduction is small (one file, four new items, two new tests) and self-contained — no external services, no flaky assumptions beyond a 120 ms timeout that comfortably brackets a 10 ms-ish first chunk and excludes a 300 ms second chunk.
The reproduction also draws a sharp line between two categories of futures that select! is happy to drop indiscriminately. The cancellation-safe sleep from step 1 is a no-op on drop; nothing observable changes. The multi-poll read_exact introduced here silently loses bytes that have already crossed the user/kernel boundary into buf, because its private progress cursor evaporates with the future state machine.
Critically, we now have a failing-by-design test we can point at. Until this step we only had passing tests with no contrast, and a tutorial about cancellation safety without a concrete failure is hand-waving. The assertion leftover == vec![0xBB; 8] is exactly the bytes-loss receipt the next steps will work to suppress.
Step 3 will pick up here: we keep the same listener, the same chunk schedule, and the same timeout, but rewrite racing_read_exact to hold an in-progress buffered read across the select! boundary using tokio::pin! and an externally owned accumulator. The leftover drain will then return zero bytes when the timeout wins, because the in-flight progress survives the next call instead of dying with the dropped future.
Repository
The state of the code after this step: 79b3ce4
Step 3: Cataloging Cancellation-Safe vs Hazardous Tokio Futures with an Executable Reference
Step 2 left us with one concrete, demonstrated hazard: a read_exact racing a sleep inside tokio::select! strands bytes when the timer wins. That single data point is enough to convince a reader the failure mode exists, but it is not enough to make decisions about every future they will ever drop into a select! arm. This step generalises the contract into a small, queryable reference: an enum that names the futures we care about, a classify function that tags each one as Safe or Unsafe, and prose attached to every variant explaining why it is on that side of the line.
Crucially, the reference is executable. The same module that exports classify also ships two #[tokio::test] demos that exercise the contract on real mpsc::Receiver and Notify instances, proving the "safe" half is not merely upstream-documented but observable in our own process. By the end of this step, anyone reading the crate can either look up a FutureKind variant or run cargo test to see the rule fire.
Setup
No new crates are needed. The tokio = { features = ["full"] } already pulled in tokio::sync, which gives us mpsc::channel and Notify for the runtime demos. The dev-dependency line from step 1 stays unchanged because start_paused = true is the only test-util surface this step touches.
All edits land in the existing src/main.rs so the file remains the single source of truth for the article. The deltas are: two new enums (CancellationSafety, FutureKind), one const fn (classify), one prose function (cancellation_safety_docs), two &[FutureKind] constants inside mod tests that name the safe/unsafe partitions, three new unit tests covering the classification table, and two new #[tokio::test] cases that demonstrate cancellation safety on mpsc::Receiver::recv and Notify::notified.
Implementation
The first addition is the binary label we will tag each future with. Keeping CancellationSafety as a two-variant enum (rather than, say, a bool) makes call sites read declaratively — Safe and Unsafe carry intent in a way that true / false never will, and we can later add a third state (e.g. ConditionallySafe) without rewriting every match arm.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CancellationSafety {
Safe,
Unsafe,
}
Next, the FutureKind enum names the thirteen tokio futures we want to classify. The list is deliberately scoped to the ones a typical service author reaches for: a timer, the channel receivers, the synchronisation primitives, and the tokio::net read/write helpers. Each variant maps to exactly one upstream future, so the reader can grep for the variant name and find a single match arm in classify and one in cancellation_safety_docs.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum FutureKind {
Sleep,
MpscRecv,
OneshotRecv,
BroadcastRecv,
WatchChanged,
NotifyNotified,
MutexLock,
SemaphoreAcquire,
TcpRead,
TcpReadExact,
TcpReadToEnd,
TcpReadLine,
TcpWriteAll,
}
The classify function is the actual rule. We mark it const so the classification participates in the type system rather than runtime dispatch — anywhere a const FOO: CancellationSafety = classify(FutureKind::Sleep); would compile, the compiler resolves the answer at build time. The body is a single match with two arms: a |-joined list of safe variants and a |-joined list of unsafe variants. Keeping it one level deep is non-negotiable per the codebase rules, but it also makes the rule auditable at a glance.
pub const fn classify(kind: FutureKind) -> CancellationSafety {
use CancellationSafety::{Safe, Unsafe};
use FutureKind::*;
match kind {
Sleep
| MpscRecv
| OneshotRecv
| BroadcastRecv
| WatchChanged
| NotifyNotified
| MutexLock
| SemaphoreAcquire
| TcpRead => Safe,
TcpReadExact | TcpReadToEnd | TcpReadLine | TcpWriteAll => Unsafe,
}
}
A boolean is not a teaching tool on its own, so cancellation_safety_docs returns a one-sentence explanation per variant. The pattern is identical across arms: state what the future holds inside its state machine, and what happens to that state on drop. For the safe variants, the note explains why dropping is observably a no-op (timers deregister, queued items stay queued, permits stay unclaimed). For the unsafe variants, the note names the partial state that cancellation strands (a per-future progress cursor, a half-filled line buffer, half a write on the wire).
pub fn cancellation_safety_docs(kind: FutureKind) -> &'static str {
match kind {
FutureKind::Sleep => {
"A timer is registered once; dropping the future deregisters \
it without altering wall-clock or runtime state."
}
FutureKind::MpscRecv => {
"Cancelling before completion does not consume a queued item: \
the next recv on the same Receiver still sees the message."
}
FutureKind::TcpReadExact => {
"Multi-poll loop that may have already pulled bytes from the \
kernel into the caller buffer; dropping mid-loop strands those bytes."
}
// ... one arm per FutureKind variant
}
}
Inside mod tests we name the two partitions as const SAFE_KINDS and const UNSAFE_KINDS slices. The classification tests then iterate each slice and assert classify agrees, which catches the failure mode where a future contributor adds a new variant but forgets to slot it into the right arm of the match — classify would refuse to compile (non-exhaustive match), but the partition tests would also refuse to pass if the contributor added the variant to the wrong slice.
const SAFE_KINDS: &[FutureKind] = &[
FutureKind::Sleep,
FutureKind::MpscRecv,
FutureKind::OneshotRecv,
FutureKind::BroadcastRecv,
FutureKind::WatchChanged,
FutureKind::NotifyNotified,
FutureKind::MutexLock,
FutureKind::SemaphoreAcquire,
FutureKind::TcpRead,
];
const UNSAFE_KINDS: &[FutureKind] = &[
FutureKind::TcpReadExact,
FutureKind::TcpReadToEnd,
FutureKind::TcpReadLine,
FutureKind::TcpWriteAll,
];
#[test]
fn classify_marks_documented_safe_futures_as_safe() {
for kind in SAFE_KINDS {
assert_eq!(classify(*kind), CancellationSafety::Safe);
}
}
#[test]
fn classify_marks_documented_unsafe_futures_as_unsafe() {
for kind in UNSAFE_KINDS {
assert_eq!(classify(*kind), CancellationSafety::Unsafe);
}
}
A third small test chains both slices through cancellation_safety_docs and asserts each returned string is non-empty. That guards against the easy regression where someone adds a FutureKind variant, slots it into classify, but forgets the docs arm — the match would still compile if a previous arm matched by accident, but it would be unreachable. We pin "every variant has a note" rather than relying on the human review.
#[test]
fn every_kind_has_a_documentation_note() {
for kind in SAFE_KINDS.iter().chain(UNSAFE_KINDS.iter()) {
let note = cancellation_safety_docs(*kind);
assert!(!note.trim().is_empty());
}
}
The two runtime demos turn the doc strings into observable behaviour. For mpsc::Receiver::recv, we open a channel with no message yet sent, race the receiver against a 50 ms sleep, and assert the receiver future did not burn the (still-pending) slot — after the select!, we send a 7, recv it back, and confirm the message arrives. start_paused = true keeps the assertion deterministic without a real timer.
#[tokio::test(start_paused = true)]
async fn mpsc_recv_does_not_consume_message_when_select_cancels_it() {
let (tx, mut rx) = mpsc::channel::<u32>(8);
tokio::select! {
v = rx.recv() => panic!("no message was sent yet, recv resolved to {v:?}"),
_ = sleep(Duration::from_millis(50)) => {}
}
tx.send(7).await.expect("send into open channel");
assert_eq!(rx.recv().await, Some(7));
}
The Notify::notified demo follows the same shape: race a never-issued notified() against a 50 ms sleep, then issue one notification after the cancellation and require a fresh notified() future to observe it within 100 ms. If Notify::notified were not cancellation-safe, the dropped future could plausibly have eaten the permit during its poll, and the second notified() would hang until the outer timeout fired.
#[tokio::test(start_paused = true)]
async fn notify_notified_does_not_burn_a_permit_when_select_cancels_it() {
let notify = Notify::new();
tokio::select! {
_ = notify.notified() => panic!("no notification was sent yet"),
_ = sleep(Duration::from_millis(50)) => {}
}
notify.notify_one();
tokio::time::timeout(Duration::from_millis(100), notify.notified())
.await
.expect("notified() should observe the permit issued after cancellation");
}
Verification
Run the full suite from the crate root. All nine tests must pass: the two sleep races from step 1, the two TCP-backed hazard tests from step 2, the three classification-table tests, and the two runtime cancellation-safety demos.
cargo test
Finished `test` profile [unoptimized + debuginfo] target(s) in 0.49s
Running unittests src/main.rs (target/debug/deps/race_demo-331142ba86e57cbe)
running 9 tests
test tests::classify_marks_documented_safe_futures_as_safe ... ok
test tests::classify_marks_documented_unsafe_futures_as_unsafe ... ok
test tests::every_kind_has_a_documentation_note ... ok
test tests::fast_branch_wins_when_its_sleep_is_shorter ... ok
test tests::slow_branch_wins_when_fast_is_longer ... ok
test tests::mpsc_recv_does_not_consume_message_when_select_cancels_it ... ok
test tests::notify_notified_does_not_burn_a_permit_when_select_cancels_it ... ok
test tests::racing_read_exact_completes_when_budget_outlasts_both_chunks ... ok
test tests::racing_read_exact_strands_bytes_when_timeout_wins ... ok
test result: ok. 9 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.61s
The two new tokio::test cases are the load-bearing ones. If a future tokio refactor regressed Notify::notified or mpsc::Receiver::recv cancellation safety, those tests would flip to failure inside the same suite that proves the read_exact hazard, and we would catch the change before propagating advice to a reader.
What we built
We now have a small, audited table covering thirteen tokio future variants split into nine cancellation-safe and four cancellation-unsafe kinds. classify is a const fn, so any caller can branch on it at compile time, and cancellation_safety_docs carries the one-sentence reason next to each variant so the reader never has to leave the crate to understand why the answer is what it is.
The classification is pinned by three unit tests that walk the partition slices, plus two tokio::test runtime demos that exercise cancellation safety on real mpsc::Receiver and Notify instances. The runtime demos are the difference between "the upstream docs claim this is safe" and "we observed it being safe inside our own process" — both matter, but only the second one fails fast when something changes.
Together with step 2's read_exact reproduction, this step closes the descriptive gap. A reader can now answer the question "is it safe to put this future in a select! arm?" by looking up the FutureKind variant, reading the one-sentence note, and (if curious) running cargo test to watch the rule hold.
Step 4 takes the table as given and starts repairing the hazard. We will reach for a buffered, pinned read that survives across select! cancellations, and use the Unsafe variants from this step's table as the catalogue of futures we need to wrap before they can be dropped safely.
Repository
The state of the code after this step: f9c205a
Step 4: Lifting Read Progress Out of the select! Future with an Owned Resumable Buffer
Step 2 produced a hard, observable failure: a read_exact racing a sleep inside tokio::select! strands the first chunk of bytes when the timer wins, and the receipt for that loss is an 8-byte leftover on the wire that the next reader has to discard. Step 3 generalised the rule into a classification table and pinned which built-in futures own progress state that dies on drop. This step takes both inputs and starts repairing the hazard.
The fix is structural rather than algorithmic: we stop trusting a single select! arm to carry multi-poll progress, and instead lift the in-progress byte cursor into a separate struct that lives outside the future. When the timer arm wins, the future is still dropped — but the buffer is not, because nothing inside select! ever owned it. The next call resumes from the saved offset, and the leftover drain that step 2 showed shrinks from eight bytes to zero.
Setup
No new crates land in this step. tokio = { features = ["full"] } already supplies AsyncReadExt::read, TcpStream, and the timer surface; everything we need is reachable from the imports step 2 introduced. The dev-dependencies row is also unchanged — the new tests reuse the existing spawn_chunked_writer helper and the same drain_with_budget shape.
All edits stay inside the existing src/main.rs. The deltas are: one new struct PersistentReadExact with five small accessors and one async fn progress method, one new free function racing_buffered_read_exact that drops the new struct into a select! arm, and four new unit tests covering the happy path, the cancellation-and-resume path, the one-shot path, and the short-stream EOF path. We keep racing_read_exact from step 2 unchanged so the hazard reproduction remains side-by-side with the fix.
Implementation
The whole repair hinges on which value owns the in-progress cursor. In step 2 the cursor lived inside the read_exact future itself, so dropping the future dropped the cursor. PersistentReadExact flips that: it owns the destination buffer and a filled count, and exposes a progress method that uses a borrowed &mut TcpStream for the duration of one poll-loop but never stores it. Because the caller holds the struct across select! iterations, no amount of cancellation can erase the bytes that were already copied in.
#[derive(Debug)]
pub struct PersistentReadExact {
buf: Vec<u8>,
filled: usize,
}
impl PersistentReadExact {
pub fn new(target_len: usize) -> Self {
Self {
buf: vec![0u8; target_len],
filled: 0,
}
}
pub fn filled(&self) -> usize {
self.filled
}
pub fn target_len(&self) -> usize {
self.buf.len()
}
pub fn is_complete(&self) -> bool {
self.filled == self.buf.len()
}
pub fn buffer(&self) -> &[u8] {
&self.buf
}
pub fn into_buffer(self) -> Vec<u8> {
self.buf
}
}
The accessors are small and boring on purpose. filled and target_len give tests a way to assert resumption progress without pattern-matching on private fields; is_complete is the single source of truth that decides whether the caller is done; buffer returns a borrowed slice for in-place inspection and into_buffer lets the caller take ownership once the read is finished. None of them touch the stream, which keeps the struct trivially Send and easy to hold across await points outside of any I/O context.
The single method that actually performs I/O is progress. It loops on stream.read into the unfilled tail of buf, advances self.filled, and either returns Ok(()) when the buffer is full or surfaces a real io::Error on a short stream. Because the loop body is the only place that mutates self.filled, every byte copied from the kernel is immediately accounted for in the externally owned struct — the very property step 2 lacked.
pub async fn progress(&mut self, stream: &mut TcpStream) -> std::io::Result<()> {
while self.filled < self.buf.len() {
let n = stream.read(&mut self.buf[self.filled..]).await?;
if n == 0 {
return Err(std::io::Error::new(
std::io::ErrorKind::UnexpectedEof,
"stream closed before buffer was filled",
));
}
self.filled += n;
}
Ok(())
}
The n == 0 branch is the small subtle invariant worth flagging. A successful AsyncReadExt::read that returns Ok(0) means the peer closed the stream cleanly with the buffer still partially filled; that is genuinely an EOF, not a spurious wake-up, so we translate it into UnexpectedEof. This matches the contract read_exact itself uses internally and keeps callers from spinning forever on a half-filled buffer.
With the struct in place, the cancellation-aware racer is a near-copy of step 2's racing_read_exact — the only change is that the read arm calls reader.progress(stream) on a borrowed &mut PersistentReadExact instead of starting a fresh read_exact against the stream. The sleep arm is byte-for-byte identical because the timer was already cancellation-safe per step 3's table.
pub async fn racing_buffered_read_exact(
stream: &mut TcpStream,
reader: &mut PersistentReadExact,
budget: Duration,
) -> ReadOutcome {
tokio::select! {
result = reader.progress(stream) => match result {
Ok(_) => ReadOutcome::Completed,
Err(_) => ReadOutcome::Failed,
},
_ = sleep(budget) => ReadOutcome::TimedOut,
}
}
The signature is what makes this work under cancellation. reader is a &mut PersistentReadExact provided by the caller, so it outlives every future the select! macro builds and tears down. When the timer arm wins, only the anonymous progress(stream) future is dropped; reader.filled retains whatever value the in-loop assignment last wrote. The next call passes the same reader in and progress simply continues from self.filled — no bytes lost, no double-read.
The first test pins the invariant a reader expects from a fresh PersistentReadExact instance: zero progress, the requested target length, an unfilled buffer of the right size, and an is_complete() answer of false. This is a #[test] rather than #[tokio::test] because the constructor performs no I/O; running it sync proves the struct's defaults do not silently rely on a runtime.
#[test]
fn fresh_persistent_reader_reports_zero_progress() {
let reader = PersistentReadExact::new(32);
assert_eq!(reader.filled(), 0);
assert_eq!(reader.target_len(), 32);
assert!(!reader.is_complete());
assert_eq!(reader.buffer().len(), 32);
}
The load-bearing test is buffered_read_preserves_progress_across_cancellation. It reuses the exact spawn_chunked_writer schedule from step 2 — two 8-byte chunks separated by a 300 ms gap — and races a 120 ms budget. On the first call the timer wins, but the assertion is that reader.filled() is 8 (chunk one), not 0. We then call racing_buffered_read_exact a second time with a generous 2-second budget and assert the buffer comes out fully filled with the concatenated chunks.
#[tokio::test]
async fn buffered_read_preserves_progress_across_cancellation() {
let addr = spawn_chunked_writer(
vec![vec![0xAA; 8], vec![0xBB; 8]],
Duration::from_millis(300),
)
.await
.expect("listener bind failed");
let mut stream = TcpStream::connect(addr)
.await
.expect("client connect failed");
let mut reader = PersistentReadExact::new(16);
let outcome = racing_buffered_read_exact(
&mut stream,
&mut reader,
Duration::from_millis(120),
)
.await;
assert_eq!(outcome, ReadOutcome::TimedOut);
assert_eq!(
reader.filled(),
8,
"chunk 1 should be retained inside the persistent buffer",
);
assert_eq!(&reader.buffer()[..8], &[0xAA; 8]);
assert!(!reader.is_complete());
let outcome = racing_buffered_read_exact(
&mut stream,
&mut reader,
Duration::from_secs(2),
)
.await;
assert_eq!(outcome, ReadOutcome::Completed);
assert!(reader.is_complete());
let expected: Vec<u8> = [vec![0xAA; 8], vec![0xBB; 8]].concat();
assert_eq!(reader.buffer(), expected.as_slice());
}
The compare-and-contrast against step 2 is the entire point: identical wire schedule, identical timeout, but the post-cancellation receipt is reader.filled() == 8 instead of leftover == vec![0xBB; 8]. The bytes that were on the wrong side of the cancellation boundary in step 2 are now on the right side of it, because the caller — not the select! future — owns the destination.
The remaining two tests close the matrix. buffered_read_completes_in_one_pass_when_budget_is_generous shows the happy path: when the budget comfortably brackets the producer schedule, a single call returns Completed with the full payload, no resume required. buffered_read_reports_eof_when_writer_closes_short shows the failure semantics for a stream that ends before the buffer is full — progress surfaces UnexpectedEof, racing_buffered_read_exact translates that into ReadOutcome::Failed, and the partial bytes that did arrive remain inspectable on reader.buffer()[..4] for the caller's own diagnostics.
#[tokio::test]
async fn buffered_read_reports_eof_when_writer_closes_short() {
let addr = spawn_chunked_writer(
vec![vec![0xAA; 4]],
Duration::from_millis(10),
)
.await
.expect("listener bind failed");
let mut stream = TcpStream::connect(addr)
.await
.expect("client connect failed");
let mut reader = PersistentReadExact::new(16);
let outcome = racing_buffered_read_exact(
&mut stream,
&mut reader,
Duration::from_secs(2),
)
.await;
assert_eq!(outcome, ReadOutcome::Failed);
assert_eq!(reader.filled(), 4);
assert_eq!(&reader.buffer()[..4], &[0xAA; 4]);
}
That EOF test matters because a callable contract has to say something useful about every terminating condition, not just success and timeout. With it in place, a caller using PersistentReadExact can distinguish three outcomes — completed, timed out, failed — and in all three cases the buffer's filled count tells them exactly how far they got.
Verification
Run the full suite from the crate root. All thirteen tests must pass: the two sleep races from step 1, the two TCP hazard tests from step 2, the three classification-table tests plus the two runtime cancellation-safety demos from step 3, and the four new buffered-read tests from this step.
cargo test
Finished `test` profile [unoptimized + debuginfo] target(s) in 0.58s
Running unittests src/main.rs (target/debug/deps/race_demo-331142ba86e57cbe)
running 13 tests
test tests::classify_marks_documented_safe_futures_as_safe ... ok
test tests::classify_marks_documented_unsafe_futures_as_unsafe ... ok
test tests::every_kind_has_a_documentation_note ... ok
test tests::fast_branch_wins_when_its_sleep_is_shorter ... ok
test tests::fresh_persistent_reader_reports_zero_progress ... ok
test tests::slow_branch_wins_when_fast_is_longer ... ok
test tests::mpsc_recv_does_not_consume_message_when_select_cancels_it ... ok
test tests::notify_notified_does_not_burn_a_permit_when_select_cancels_it ... ok
test tests::buffered_read_reports_eof_when_writer_closes_short ... ok
test tests::buffered_read_completes_in_one_pass_when_budget_is_generous ... ok
test tests::racing_read_exact_completes_when_budget_outlasts_both_chunks ... ok
test tests::racing_read_exact_strands_bytes_when_timeout_wins ... ok
test tests::buffered_read_preserves_progress_across_cancellation ... ok
test result: ok. 13 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.73s
The two tests doing the diagnostic work are racing_read_exact_strands_bytes_when_timeout_wins (the unrepaired hazard, kept on purpose) and buffered_read_preserves_progress_across_cancellation (the repair). Together they form a side-by-side pair: drop them both into the same suite, and any future regression on either side surfaces immediately rather than hiding behind a successful overall run.
What we built
We added an externally owned PersistentReadExact buffer plus a racing_buffered_read_exact helper that drops it into a tokio::select! arm without paying the cancellation tax. The struct's progress method holds its in-flight cursor on self, not inside the anonymous future, so a timer-cancellation tears down only the future and leaves the bytes intact for the next call.
The fix is a single shape that generalises to other multi-poll I/O helpers from step 3's Unsafe list. Whenever you can move the progress state out of the future and onto a struct the caller owns across iterations, the future becomes cheap to drop and the read becomes resumable. read_to_end would use a Vec<u8> accumulator the caller holds; read_line would use a String plus a "saw a carriage return" flag; write_all would use a &[u8] slice cursor — same recipe, different state.
We also wrote a test pair that turns the repair into a tracked invariant. The hazard reproduction from step 2 still runs and still fails the way it failed before — that is the regression alarm for the problem. The new buffered-read test passes today and asserts the byte-perfect resume — that is the regression alarm for the fix. Either side flipping in CI is a meaningful, isolable signal.
Step 5 will keep PersistentReadExact exactly as it is and turn its attention to the select! macro itself. We will switch from the default fair (randomised) poll order to biased; so the read arm is always polled before the timer arm, giving deterministic ordering under load and laying the groundwork for the shutdown-signal pattern in step 6.
Repository
The state of the code after this step: 83795cb
Step 5: Pinning Poll Order with biased; so the Read Arm Wins Ties Over the Timer
Step 4 closed the byte-stranding hole by lifting the read cursor onto a caller-owned PersistentReadExact, so a timer-cancellation no longer destroys progress. What it did not address is the order in which tokio::select! polls its arms: by default the macro picks a random starting index per call, which means a chunk that is sitting in the kernel buffer can lose the race to a timer that fired the same instant the wake-up arrived. For a buffered reader that already knows how to resume this is harmless, but for a priority signal — "always prefer shutdown over normal work" — randomised order is the wrong default.
This step makes one local edit and one new demo. The local edit adds biased; as the first line of the racing_buffered_read_exact macro body so the read is always polled first. The demo introduces a small two-channel priority picker — one biased, one fair — and a test pair that proves the biased version is deterministic over hundreds of iterations where the fair version distributes wins across both branches.
Setup
No new crates are added. tokio = { features = ["full"] } already covers tokio::sync::mpsc, the Receiver::recv future, and the existing sleep import; the runtime already has test-util in dev-dependencies. The whole change is a single-file diff against src/main.rs, with no module reorganisation.
The deliverables are: one keyword (biased;) added to the existing racing_buffered_read_exact macro body, one new public enum PriorityOutcome, two new free functions biased_priority_pick and fair_priority_pick that share an identical signature so they can be A/B tested, and three new unit tests. We also keep an iterations constant at the test-module level so the fair-vs-biased experiment runs enough times that randomness would visibly show up in the wrong branch winning.
Implementation
The first change is the smallest one in the entire article — a single line. We add biased; as the first statement inside the tokio::select! body of racing_buffered_read_exact so the macro stops randomising the starting index and instead polls the read arm before the timer arm on every call.
pub async fn racing_buffered_read_exact(
stream: &mut TcpStream,
reader: &mut PersistentReadExact,
budget: Duration,
) -> ReadOutcome {
tokio::select! {
biased;
result = reader.progress(stream) => match result {
Ok(_) => ReadOutcome::Completed,
Err(_) => ReadOutcome::Failed,
},
_ = sleep(budget) => ReadOutcome::TimedOut,
}
}
The semantic shift is subtle but important. Without biased;, the macro generates a random permutation of arm indices on every poll and calls each branch's future in that order; with biased;, the order is exactly the source order, top to bottom. For the read arm that is sitting on already-arrived bytes, that means the read is observed Ready and consumed on the very same poll the wake-up notification arrived on, instead of being deferred to a subsequent poll because the timer happened to be drawn first.
The second change is a self-contained priority demo. It exists to make the difference between biased and the default fair behaviour visible without involving TCP sockets, because the property we are asserting is purely about macro-level branch selection, not about I/O readiness.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PriorityOutcome {
HighPriority,
LowPriority,
}
pub async fn biased_priority_pick(
high: &mut tokio::sync::mpsc::Receiver<()>,
low: &mut tokio::sync::mpsc::Receiver<()>,
) -> PriorityOutcome {
tokio::select! {
biased;
Some(_) = high.recv() => PriorityOutcome::HighPriority,
Some(_) = low.recv() => PriorityOutcome::LowPriority,
}
}
pub async fn fair_priority_pick(
high: &mut tokio::sync::mpsc::Receiver<()>,
low: &mut tokio::sync::mpsc::Receiver<()>,
) -> PriorityOutcome {
tokio::select! {
Some(_) = high.recv() => PriorityOutcome::HighPriority,
Some(_) = low.recv() => PriorityOutcome::LowPriority,
}
}
The two functions are byte-for-byte identical except for the biased; directive, which is exactly what makes them a fair A/B test. Both receive &mut borrows of their channels so the caller keeps ownership of state across iterations, and both pattern-match Some(_) because we are interested in a delivered value rather than channel closure semantics.
The first test asserts the determinism property. It pre-loads one message into each channel — so both arms are immediately Ready when select! first polls — and then runs the biased picker PRIORITY_ITERATIONS times in a row. The assertion is that every single iteration returns HighPriority, with the iteration index surfaced in the panic message so a future regression points at the exact run that broke.
const PRIORITY_ITERATIONS: usize = 200;
async fn preload_both_channels() -> (mpsc::Receiver<()>, mpsc::Receiver<()>) {
let (high_tx, high_rx) = mpsc::channel::<()>(1);
let (low_tx, low_rx) = mpsc::channel::<()>(1);
high_tx.send(()).await.expect("high preload");
low_tx.send(()).await.expect("low preload");
(high_rx, low_rx)
}
#[tokio::test]
async fn biased_select_always_picks_high_priority_when_both_are_ready() {
for iteration in 0..PRIORITY_ITERATIONS {
let (mut high_rx, mut low_rx) = preload_both_channels().await;
let outcome = biased_priority_pick(&mut high_rx, &mut low_rx).await;
assert_eq!(
outcome,
PriorityOutcome::HighPriority,
"biased select! must pick the first listed branch every time \
(iteration {iteration} picked {outcome:?})",
);
}
}
Two hundred iterations is comfortably above the threshold where a uniformly random pick would be expected to land on LowPriority at least once — the probability of two hundred consecutive HighPriority results under a fair coin is roughly 2.5e-60, so any flake on this test is a real regression rather than statistical noise. The preload_both_channels helper is split out so the second test can reuse it without duplicating the setup.
The second test asserts the negative — that the default fair select! actually does distribute wins across both branches over the same number of iterations. This is the test that protects the demo: if the fair picker ever silently stops being fair (a future tokio change, an accidental biased; slipped in), the A/B comparison collapses and the biased story stops being a contrast.
#[tokio::test]
async fn fair_select_eventually_picks_low_priority_when_both_are_ready() {
let mut high_wins = 0usize;
let mut low_wins = 0usize;
for _ in 0..PRIORITY_ITERATIONS {
let (mut high_rx, mut low_rx) = preload_both_channels().await;
match fair_priority_pick(&mut high_rx, &mut low_rx).await {
PriorityOutcome::HighPriority => high_wins += 1,
PriorityOutcome::LowPriority => low_wins += 1,
}
}
assert!(
high_wins > 0 && low_wins > 0,
"default select! must distribute wins across branches when both are ready \
(high={high_wins}, low={low_wins})",
);
}
The assertion is deliberately weak — "at least one win per side over 200 trials" — because the goal is to confirm distribution exists, not to pin the exact ratio. Pinning the ratio would tie the test to internal tokio implementation details that are explicitly documented as unspecified, and would make the suite brittle across runtime upgrades.
The third test is a regression guard for the local edit. After adding biased; to racing_buffered_read_exact, we want to confirm the change did not accidentally break the existing happy path when only the read arm is ready. It re-runs the same generous-budget scenario from step 4 and asserts the buffer still comes out fully filled.
#[tokio::test]
async fn biased_buffered_read_still_completes_when_only_read_is_ready() {
let addr = spawn_chunked_writer(
vec![vec![0xAA; 8], vec![0xBB; 8]],
Duration::from_millis(20),
)
.await
.expect("listener bind failed");
let mut stream = TcpStream::connect(addr)
.await
.expect("client connect failed");
let mut reader = PersistentReadExact::new(16);
let outcome = racing_buffered_read_exact(
&mut stream,
&mut reader,
Duration::from_secs(2),
)
.await;
assert_eq!(outcome, ReadOutcome::Completed);
let expected: Vec<u8> = [vec![0xAA; 8], vec![0xBB; 8]].concat();
assert_eq!(reader.buffer(), expected.as_slice());
}
This is the test we will rely on later when chaining a shutdown-signal arm in step 6: if priority-driven select! arms ever start starving the read, this test will turn red before any downstream user code does.
Verification
Run the suite from the crate root. The new test count is sixteen — thirteen carried over from step 4 plus the three priority-ordering tests added in this step.
cargo test
Finished `test` profile [unoptimized + debuginfo] target(s) in 0.61s
Running unittests src/main.rs (target/debug/deps/race_demo-331142ba86e57cbe)
running 16 tests
test tests::classify_marks_documented_safe_futures_as_safe ... ok
test tests::classify_marks_documented_unsafe_futures_as_unsafe ... ok
test tests::every_kind_has_a_documentation_note ... ok
test tests::fast_branch_wins_when_its_sleep_is_shorter ... ok
test tests::fresh_persistent_reader_reports_zero_progress ... ok
test tests::slow_branch_wins_when_fast_is_longer ... ok
test tests::mpsc_recv_does_not_consume_message_when_select_cancels_it ... ok
test tests::notify_notified_does_not_burn_a_permit_when_select_cancels_it ... ok
test tests::buffered_read_reports_eof_when_writer_closes_short ... ok
test tests::buffered_read_completes_in_one_pass_when_budget_is_generous ... ok
test tests::racing_read_exact_completes_when_budget_outlasts_both_chunks ... ok
test tests::biased_buffered_read_still_completes_when_only_read_is_ready ... ok
test tests::racing_read_exact_strands_bytes_when_timeout_wins ... ok
test tests::buffered_read_preserves_progress_across_cancellation ... ok
test tests::biased_select_always_picks_high_priority_when_both_are_ready ... ok
test tests::fair_select_eventually_picks_low_priority_when_both_are_ready ... ok
test result: ok. 16 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.81s
The three new tests are the ones doing the diagnostic work: biased_select_always_picks_high_priority_when_both_are_ready proves the new determinism, fair_select_eventually_picks_low_priority_when_both_are_ready proves the default contrast still holds, and biased_buffered_read_still_completes_when_only_read_is_ready proves the local edit to racing_buffered_read_exact did not regress the happy path.
What we built
We added the biased; directive to the buffered read racer and a separate two-channel demo that makes the macro-level ordering property testable in isolation. The directive replaces tokio's randomised arm-permutation with a strict top-down order, which means a branch that wants priority simply needs to be listed first.
The priority demo is a pattern we will reach for again in step 6. The shape — two mpsc::Receiver borrows held across select! calls, biased on the shutdown side — is the canonical Tokio idiom for "respond to ctrl-c immediately, even if there is also work pending in the queue". By proving the determinism with 200 trials per run, the test suite turns the macro keyword into a tracked invariant rather than a hopeful comment.
The three new tests also give us a regression matrix with three distinct failure modes. biased_select_always_picks_high_priority_when_both_are_ready catches an accidental removal of the biased; keyword; fair_select_eventually_picks_low_priority_when_both_are_ready catches an over-eager addition of biased; to the wrong arm; biased_buffered_read_still_completes_when_only_read_is_ready catches a starvation regression where biasing the read arm somehow blocks the timer arm from ever firing.
Step 6 will compose biased; with a shutdown-signal Notify to build a graceful-shutdown wrapper around the buffered reader. The biased ordering established here is the prerequisite — without deterministic priority, "respond to shutdown before continuing the read" would be a 50/50 coin flip per poll.
Repository
The state of the code after this step: fa7d094
Step 6: Composing a Shutdown-First Worker Loop on Top of biased; Select Priority
Step 5 pinned the poll order inside tokio::select! with the biased; directive and proved the property in isolation with a two-channel A/B test. What it did not yet do is wire that property into a realistic loop, where the same select! body runs over and over and the priority branch has to win on every iteration, not just once. A worker that pulls jobs off an mpsc queue and watches for a shutdown signal is the canonical place this matters: if a shutdown arrives while there are still buffered jobs, the worker must stop before dequeuing the next item, otherwise graceful shutdown turns into "process whatever happened to be visible this poll, then exit."
This step assembles that loop. We add a run_biased_worker function that loops on a biased two-arm select!, a run_fair_worker control that is byte-identical except for the missing biased;, and a four-test matrix that proves the biased version exits deterministically on shutdown while the fair version distributes its outcomes across both branches. The same channel-preloading trick from step 5 lets us run the property hundreds of times per test, so the determinism claim is enforced rather than asserted.
Setup
No new crates are added. We already depend on tokio with the full feature flag, which gives us tokio::sync::mpsc::Receiver for the work queue and tokio::sync::watch::Receiver for the shutdown signal. The whole step is a single-file diff against src/main.rs, with no module reorganisation and no new dev-dependencies.
The deliverables are: two new types (WorkerExit and WorkerReport), one biased worker function, one fair-control worker function, three small private helpers that keep nesting flat (drain_pending, shutdown_report, closed_report), and four new unit tests. We also keep WORKER_TRIALS and PRELOADED_WORK constants at the test-module level so the determinism experiment runs enough times that any priority leak would show up as a hard failure rather than a flaky one.
Implementation
The first new type is the exit-reason enum. The worker can stop for exactly two reasons — a shutdown signal was raised, or the work channel was closed by all senders — and we want the test suite to assert which reason fired, not just that the loop returned.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum WorkerExit {
ShutdownRequested,
WorkChannelClosed,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WorkerReport {
pub processed: Vec<u64>,
pub remaining_after_shutdown: usize,
pub exit_reason: WorkerExit,
}
WorkerReport is a value-return struct rather than a tuple because the three fields each answer a distinct question: what got processed, what was still queued at the moment of shutdown, and why the loop exited. Keeping them named makes the tests read like assertions about behaviour rather than positional unpacks.
The biased worker is the centrepiece. It loops over a two-arm select! with biased; as the first directive, so the shutdown arm is always polled before the work arm. On shutdown, it drains whatever is still buffered in the work channel using a non-blocking try_recv loop, so the caller can see exactly how much was left behind.
pub async fn run_biased_worker(
work: &mut tokio::sync::mpsc::Receiver<u64>,
shutdown: &mut tokio::sync::watch::Receiver<bool>,
) -> WorkerReport {
let mut processed = Vec::new();
loop {
tokio::select! {
biased;
_ = shutdown.changed() => {
return shutdown_report(processed, drain_pending(work));
}
maybe = work.recv() => match maybe {
Some(item) => processed.push(item),
None => return closed_report(processed),
},
}
}
}
The shape is deliberate. Both receivers are passed as &mut so the caller keeps ownership across the call and can reuse the channels after the worker returns. The work arm matches both Some(item) and None rather than using the Some(item) => pattern guard, so a closed work channel is a clean exit with a different WorkerExit variant rather than a panic on missing pattern. The shutdown arm ignores the Result from changed() — once shutdown is requested, the worker stops regardless of whether the publisher dropped cleanly or not.
The fair-control worker is byte-for-byte identical except for the missing biased; line. It exists only so the test suite has a head-to-head comparison: any drift in tokio's default fairness model, or any accidental copy-paste that leaves biased; out of the real worker, has to flip exactly one test rather than going unnoticed.
pub async fn run_fair_worker(
work: &mut tokio::sync::mpsc::Receiver<u64>,
shutdown: &mut tokio::sync::watch::Receiver<bool>,
) -> WorkerReport {
let mut processed = Vec::new();
loop {
tokio::select! {
_ = shutdown.changed() => {
return shutdown_report(processed, drain_pending(work));
}
maybe = work.recv() => match maybe {
Some(item) => processed.push(item),
None => return closed_report(processed),
},
}
}
}
Three small helpers keep the select! arms flat. The codebase rules cap if/else nesting at two levels and forbid nested error handling, so the report-construction logic lives in named functions instead of inline blocks. drain_pending uses try_recv in a while loop because we want to count the survivors without waiting for more, and we want the count to be accurate even if a sender drops mid-call.
fn drain_pending(work: &mut tokio::sync::mpsc::Receiver<u64>) -> usize {
let mut count = 0;
while work.try_recv().is_ok() {
count += 1;
}
count
}
fn shutdown_report(processed: Vec<u64>, remaining: usize) -> WorkerReport {
WorkerReport {
processed,
remaining_after_shutdown: remaining,
exit_reason: WorkerExit::ShutdownRequested,
}
}
fn closed_report(processed: Vec<u64>) -> WorkerReport {
WorkerReport {
processed,
remaining_after_shutdown: 0,
exit_reason: WorkerExit::WorkChannelClosed,
}
}
The test suite mirrors the structure of step 5: a setup helper that puts both branches into a "simultaneously ready" state, then a determinism test, a regression-guard, a happy-path test, and a fairness-control test. The setup helper preloads the work queue with ten items and publishes the shutdown signal before returning, so when the worker is awaited the very first poll sees both branches as ready.
const WORKER_TRIALS: usize = 200;
const PRELOADED_WORK: u64 = 10;
async fn preload_and_signal_shutdown(
items: u64,
) -> (mpsc::Receiver<u64>, watch::Receiver<bool>) {
let (work_tx, work_rx) = mpsc::channel::<u64>(64);
let (shutdown_tx, shutdown_rx) = watch::channel(false);
for i in 0..items {
work_tx.send(i).await.expect("preload work item");
}
shutdown_tx.send(true).expect("publish shutdown signal");
(work_rx, shutdown_rx)
}
The first test asserts the headline property. The biased worker must observe ShutdownRequested before dequeuing any item, and all ten preloaded items must still be drainable from the remaining count. This is the test that turns "biased select chooses shutdown" from a hopeful comment into a tracked invariant.
#[tokio::test]
async fn biased_worker_exits_on_shutdown_without_consuming_buffered_work() {
let (mut work_rx, mut shutdown_rx) =
preload_and_signal_shutdown(PRELOADED_WORK).await;
let report = run_biased_worker(&mut work_rx, &mut shutdown_rx).await;
assert_eq!(report.exit_reason, WorkerExit::ShutdownRequested);
assert!(
report.processed.is_empty(),
"biased worker must observe shutdown before any item is dequeued, \
processed instead: {:?}",
report.processed,
);
assert_eq!(
report.remaining_after_shutdown,
PRELOADED_WORK as usize,
"all preloaded items must still be drainable after the biased exit",
);
}
The second test re-runs the same scenario WORKER_TRIALS times in a row. Two hundred consecutive shutdown-wins under a uniform random pick would have probability 2.5e-60, so a single trial that processes any item under the biased worker is a real regression rather than statistical noise. We surface the trial index in the panic message so a future failure points at the exact run that broke.
#[tokio::test]
async fn biased_worker_is_deterministic_across_many_preempted_runs() {
for trial in 0..WORKER_TRIALS {
let (mut work_rx, mut shutdown_rx) =
preload_and_signal_shutdown(PRELOADED_WORK).await;
let report = run_biased_worker(&mut work_rx, &mut shutdown_rx).await;
assert_eq!(
report.exit_reason,
WorkerExit::ShutdownRequested,
"trial {trial} exited with the wrong reason: {:?}",
report.exit_reason,
);
assert!(
report.processed.is_empty(),
"trial {trial} processed items before shutdown: {:?}",
report.processed,
);
}
}
The third test is the happy-path regression: if no shutdown ever arrives, the biased worker must still drain the entire work queue and exit cleanly when the senders close. This is the test that catches a starvation regression where prioritising the shutdown arm accidentally blocks the work arm from ever firing.
#[tokio::test]
async fn biased_worker_drains_buffered_work_to_completion_when_no_shutdown_arrives() {
let (work_tx, mut work_rx) = mpsc::channel::<u64>(64);
let (_shutdown_tx, mut shutdown_rx) = watch::channel(false);
for i in 0..5u64 {
work_tx.send(i).await.expect("send work item");
}
drop(work_tx);
let report = run_biased_worker(&mut work_rx, &mut shutdown_rx).await;
assert_eq!(report.exit_reason, WorkerExit::WorkChannelClosed);
assert_eq!(report.processed, vec![0, 1, 2, 3, 4]);
assert_eq!(report.remaining_after_shutdown, 0);
}
The fourth test is the fair-control. Across the same 200 preloaded-and-shutdown trials, the default (non-biased) worker must produce both outcomes: at least one trial where zero items were processed (shutdown won the first poll), and at least one trial where some items were processed (work won the first poll). The assertion is intentionally weak — "both buckets are non-empty" — because pinning a ratio would tie the test to internal tokio fairness internals that are documented as unspecified.
#[tokio::test]
async fn fair_worker_distributes_wins_so_some_runs_process_items_before_shutdown() {
let mut zero_processed = 0usize;
let mut some_processed = 0usize;
for _ in 0..WORKER_TRIALS {
let (mut work_rx, mut shutdown_rx) =
preload_and_signal_shutdown(PRELOADED_WORK).await;
let report = run_fair_worker(&mut work_rx, &mut shutdown_rx).await;
assert_eq!(report.exit_reason, WorkerExit::ShutdownRequested);
if report.processed.is_empty() {
zero_processed += 1;
} else {
some_processed += 1;
}
assert_eq!(
report.processed.len() + report.remaining_after_shutdown,
PRELOADED_WORK as usize,
"every item should be accounted for either as processed or drained",
);
}
assert!(
zero_processed > 0 && some_processed > 0,
"default select! must distribute wins across branches \
(zero={zero_processed}, some={some_processed} out of {WORKER_TRIALS})",
);
}
The accounting assertion processed.len() + remaining_after_shutdown == PRELOADED_WORK is the other invariant we want bolted down: no matter which arm wins, every preloaded item must be accounted for exactly once, either as already-processed or as still-queued. A leak in either direction would show up here before it reaches a real consumer.
Verification
Run the suite from the crate root. The new test count is twenty — sixteen carried over from step 5 plus the four shutdown-loop tests added in this step.
cargo test
Finished `test` profile [unoptimized + debuginfo] target(s) in 0.21s
Running unittests src/main.rs (target/debug/deps/race_demo-331142ba86e57cbe)
running 20 tests
test tests::classify_marks_documented_safe_futures_as_safe ... ok
test tests::classify_marks_documented_unsafe_futures_as_unsafe ... ok
test tests::every_kind_has_a_documentation_note ... ok
test tests::fast_branch_wins_when_its_sleep_is_shorter ... ok
test tests::biased_worker_drains_buffered_work_to_completion_when_no_shutdown_arrives ... ok
test tests::biased_worker_exits_on_shutdown_without_consuming_buffered_work ... ok
test tests::buffered_read_reports_eof_when_writer_closes_short ... ok
test tests::fresh_persistent_reader_reports_zero_progress ... ok
test tests::notify_notified_does_not_burn_a_permit_when_select_cancels_it ... ok
test tests::slow_branch_wins_when_fast_is_longer ... ok
test tests::mpsc_recv_does_not_consume_message_when_select_cancels_it ... ok
test tests::fair_select_eventually_picks_low_priority_when_both_are_ready ... ok
test tests::biased_select_always_picks_high_priority_when_both_are_ready ... ok
test tests::fair_worker_distributes_wins_so_some_runs_process_items_before_shutdown ... ok
test tests::biased_worker_is_deterministic_across_many_preempted_runs ... ok
test tests::biased_buffered_read_still_completes_when_only_read_is_ready ... ok
test tests::buffered_read_completes_in_one_pass_when_budget_is_generous ... ok
test tests::racing_read_exact_completes_when_budget_outlasts_both_chunks ... ok
test tests::buffered_read_preserves_progress_across_cancellation ... ok
test tests::racing_read_exact_strands_bytes_when_timeout_wins ... ok
test result: ok. 20 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.31s
The four new tests are the ones doing the work for this step. biased_worker_exits_on_shutdown_without_consuming_buffered_work proves the single-shot property, biased_worker_is_deterministic_across_many_preempted_runs proves it holds across 200 trials, biased_worker_drains_buffered_work_to_completion_when_no_shutdown_arrives guards the happy path against starvation, and fair_worker_distributes_wins_so_some_runs_process_items_before_shutdown keeps the contrast story honest by failing if the non-biased control ever silently stops being fair.
What we built
We composed the biased; primitive from step 5 into a real worker loop. The loop awaits a watch::Receiver<bool> shutdown channel and an mpsc::Receiver<u64> work channel inside a top-down select!, so a shutdown signal that arrives during a poll always wins the same poll — the worker never dequeues another item once shutdown is observed.
The four-test matrix turns the priority guarantee into a tracked invariant. biased_worker_exits_on_shutdown_without_consuming_buffered_work is the single-shot proof, biased_worker_is_deterministic_across_many_preempted_runs lifts it to a 200-trial property check, biased_worker_drains_buffered_work_to_completion_when_no_shutdown_arrives proves the read arm still drains when no shutdown ever fires, and fair_worker_distributes_wins_so_some_runs_process_items_before_shutdown keeps the comparison honest by asserting the default fair behaviour actually distributes wins.
The WorkerReport shape is the other deliverable worth calling out. By returning processed, remaining_after_shutdown, and exit_reason as named fields, the loop hands back enough information for the caller to drive a graceful drain phase, log a shutdown summary, or decide whether to retry — without having to inspect the channels after the fact.
This is the pattern most real Tokio services should reach for when they handle SIGINT. Swap the watch::Receiver<bool> for a tokio::signal::ctrl_c() handler or a broadcast channel and the priority story is identical: shutdown listed first, biased directive at the top of the body, named report on the way out.
Repository
The state of the code after this step: f9793b9
Step 7: Microbenchmarking Fair vs biased; select! Under Load and Pinning the Trade-offs as Tests
Step 6 composed the biased; directive into a real shutdown-first worker loop and proved the priority guarantee across 200 trials. What it left unsaid is what you give up in exchange — the fair default is not just a different policy, it is the policy you want every time a starved branch would be a bug. Up to this point the article has only argued that contrast with prose; in this step we measure it.
We add a small in-process microbenchmark that pushes both biased_priority_pick and fair_priority_pick through a tight loop where both branches are always ready, records the per-iteration timing alongside the per-branch win counts, and then folds the two results into a single TradeoffSummary value that the test suite can assert against. Crucially, the summary is shaped so a regression in either direction — biased select starting to "fairly" let the low branch through, or fair select collapsing to a deterministic pick — flips at least one test rather than slipping into a chart nobody reads.
Setup
No new crates are added and no new files are created. The benchmark and its supporting types live alongside the rest of the worker code in src/main.rs, and the new tests join the existing #[cfg(test)] mod tests block. The only standard-library change is bringing std::time::Instant into scope at the top of the file, next to the existing Duration import, so the loop can capture an elapsed wall-clock measurement without pulling in a benchmark harness.
The deliverables are: a PriorityLoadBenchmark struct with three derived metrics (iterations_per_sec, mean_iter_nanos, first_branch_share), two driver functions (run_biased_priority_load and run_fair_priority_load), a TradeoffSummary value with a summarize_tradeoff reducer, and four new unit tests. We deliberately reuse biased_priority_pick and fair_priority_pick from step 5 rather than inlining a fresh select! body — the whole point is to measure the same primitives the rest of the article has been describing.
Implementation
The benchmark record carries the raw numbers (iteration count, elapsed time, two branch-win counters) and lets the test layer derive whatever rate it wants. Methods, not fields, expose the rates so the divide-by-zero guards live in one place.
#[derive(Debug, Clone, PartialEq)]
pub struct PriorityLoadBenchmark {
pub iterations: usize,
pub elapsed: Duration,
pub first_branch_wins: usize,
pub second_branch_wins: usize,
}
impl PriorityLoadBenchmark {
pub fn iterations_per_sec(&self) -> f64 {
let secs = self.elapsed.as_secs_f64();
if secs <= 0.0 {
return 0.0;
}
self.iterations as f64 / secs
}
pub fn mean_iter_nanos(&self) -> f64 {
if self.iterations == 0 {
return 0.0;
}
self.elapsed.as_nanos() as f64 / self.iterations as f64
}
pub fn first_branch_share(&self) -> f64 {
if self.iterations == 0 {
return 0.0;
}
self.first_branch_wins as f64 / self.iterations as f64
}
}
Every derived metric returns 0.0 for the empty case instead of panicking. That sounds finicky for a benchmark, but a zero-iteration PriorityLoadBenchmark is exactly the value a future caller might construct to represent "no run took place" — making the metrics total avoids a NaN leaking into the TradeoffSummary and turning a clean test into a confusing failure later.
The driver functions are intentionally symmetric: same channel setup, same accounting, only the select! flavour differs. A small helper builds the "both branches ready" state by preloading a one-slot mpsc on each side, which is the exact condition where biased; and the fair default disagree.
async fn loaded_priority_channels(
) -> (tokio::sync::mpsc::Sender<()>, tokio::sync::mpsc::Receiver<()>, tokio::sync::mpsc::Sender<()>, tokio::sync::mpsc::Receiver<()>)
{
let (high_tx, high_rx) = tokio::sync::mpsc::channel::<()>(1);
let (low_tx, low_rx) = tokio::sync::mpsc::channel::<()>(1);
high_tx.send(()).await.expect("preload high");
low_tx.send(()).await.expect("preload low");
(high_tx, high_rx, low_tx, low_rx)
}
fn record_outcome(
outcome: PriorityOutcome,
first_branch_wins: &mut usize,
second_branch_wins: &mut usize,
) {
match outcome {
PriorityOutcome::HighPriority => *first_branch_wins += 1,
PriorityOutcome::LowPriority => *second_branch_wins += 1,
}
}
record_outcome lives outside the driver loops so the bookkeeping is shared and the loop bodies stay flat — a single match per iteration with no nested if let. The senders are returned from loaded_priority_channels and bound to _h_tx / _l_tx in the driver so the channels stay open for the full iteration; dropping them would close the receivers between the preload and the select!, defeating the "both ready" invariant.
pub async fn run_biased_priority_load(iterations: usize) -> PriorityLoadBenchmark {
let mut first_branch_wins = 0;
let mut second_branch_wins = 0;
let start = Instant::now();
for _ in 0..iterations {
let (_h_tx, mut high_rx, _l_tx, mut low_rx) = loaded_priority_channels().await;
let outcome = biased_priority_pick(&mut high_rx, &mut low_rx).await;
record_outcome(outcome, &mut first_branch_wins, &mut second_branch_wins);
}
let elapsed = start.elapsed();
PriorityLoadBenchmark {
iterations,
elapsed,
first_branch_wins,
second_branch_wins,
}
}
run_fair_priority_load is the byte-identical mirror that swaps in fair_priority_pick. Keeping both functions in the file means the test below can run them back-to-back on the same async runtime and compare their outputs directly, without the noise of separate processes or harness warm-up.
The final piece is the trade-off reducer. It turns the raw first_branch_share floats into rounded percentages (so test assertions can compare integers, never floats with assert_eq!) and flags the two qualitative properties we actually care about: did the biased run starve the second branch, and did the fair run avoid that fate?
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct TradeoffSummary {
pub biased_first_branch_share_pct: u32,
pub fair_first_branch_share_pct: u32,
pub biased_starves_second_branch: bool,
pub fair_starves_second_branch: bool,
}
pub fn summarize_tradeoff(
biased: &PriorityLoadBenchmark,
fair: &PriorityLoadBenchmark,
) -> TradeoffSummary {
TradeoffSummary {
biased_first_branch_share_pct: to_percent(biased.first_branch_share()),
fair_first_branch_share_pct: to_percent(fair.first_branch_share()),
biased_starves_second_branch: biased.second_branch_wins == 0,
fair_starves_second_branch: fair.second_branch_wins == 0,
}
}
fn to_percent(share: f64) -> u32 {
(share * 100.0).round() as u32
}
The four new tests then turn the benchmark from a "look at the numbers" experiment into tracked invariants. The biased test asserts a 100% first-branch share under load; the fair test asserts both branches win at least once; the summary test asserts the contrast survives end-to-end; and a fourth test guards the empty-benchmark divisor case so the metrics never leak NaN.
const BENCHMARK_ITERATIONS: usize = 500;
const BENCHMARK_BUDGET: Duration = Duration::from_secs(10);
#[tokio::test]
async fn biased_load_benchmark_always_picks_first_branch_within_budget() {
let result = run_biased_priority_load(BENCHMARK_ITERATIONS).await;
assert_eq!(result.iterations, BENCHMARK_ITERATIONS);
assert_eq!(
result.first_branch_wins, BENCHMARK_ITERATIONS,
"biased select! must pick the first branch on every iteration"
);
assert_eq!(result.second_branch_wins, 0);
assert!(result.elapsed < BENCHMARK_BUDGET);
assert!(result.iterations_per_sec() > 0.0);
assert!(result.mean_iter_nanos() > 0.0);
}
The BENCHMARK_BUDGET of ten seconds is generous enough that a slow CI runner won't flake but tight enough that a real regression — say, a future tokio version adding a per-poll syscall — would blow past it. The other two assertions on iterations_per_sec and mean_iter_nanos exist to make sure the metric helpers themselves don't silently start returning zero.
Verification
Run the crate's test suite from the codebase root. The total test count is now 24 — twenty carried over from step 6 plus the four benchmark tests added here.
cargo test
Finished `test` profile [unoptimized + debuginfo] target(s) in 0.54s
Running unittests src/main.rs (target/debug/deps/race_demo-331142ba86e57cbe)
running 24 tests
test tests::classify_marks_documented_safe_futures_as_safe ... ok
test tests::classify_marks_documented_unsafe_futures_as_unsafe ... ok
test tests::empty_benchmark_returns_zero_metrics_instead_of_dividing_by_zero ... ok
test tests::every_kind_has_a_documentation_note ... ok
test tests::biased_worker_exits_on_shutdown_without_consuming_buffered_work ... ok
test tests::biased_worker_drains_buffered_work_to_completion_when_no_shutdown_arrives ... ok
test tests::fast_branch_wins_when_its_sleep_is_shorter ... ok
test tests::buffered_read_reports_eof_when_writer_closes_short ... ok
test tests::fresh_persistent_reader_reports_zero_progress ... ok
test tests::notify_notified_does_not_burn_a_permit_when_select_cancels_it ... ok
test tests::mpsc_recv_does_not_consume_message_when_select_cancels_it ... ok
test tests::fair_select_eventually_picks_low_priority_when_both_are_ready ... ok
test tests::biased_select_always_picks_high_priority_when_both_are_ready ... ok
test tests::slow_branch_wins_when_fast_is_longer ... ok
test tests::fair_load_benchmark_distributes_wins_across_both_branches ... ok
test tests::biased_load_benchmark_always_picks_first_branch_within_budget ... ok
test tests::biased_worker_is_deterministic_across_many_preempted_runs ... ok
test tests::fair_worker_distributes_wins_so_some_runs_process_items_before_shutdown ... ok
test tests::tradeoff_summary_captures_priority_vs_distribution_contrast ... ok
test tests::biased_buffered_read_still_completes_when_only_read_is_ready ... ok
test tests::buffered_read_completes_in_one_pass_when_budget_is_generous ... ok
test tests::racing_read_exact_completes_when_budget_outlasts_both_chunks ... ok
test tests::buffered_read_preserves_progress_across_cancellation ... ok
test tests::racing_read_exact_strands_bytes_when_timeout_wins ... ok
test result: ok. 24 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.31s
The four new tests doing the work here are biased_load_benchmark_always_picks_first_branch_within_budget, fair_load_benchmark_distributes_wins_across_both_branches, tradeoff_summary_captures_priority_vs_distribution_contrast, and empty_benchmark_returns_zero_metrics_instead_of_dividing_by_zero. Together they pin the priority-vs-distribution trade-off down to numbers that fail loudly if either policy quietly drifts.
What we built
We turned the qualitative argument from steps 5 and 6 — "biased gives deterministic priority, fair distributes wins" — into a quantitative experiment encoded as four passing tests. The PriorityLoadBenchmark value carries the raw iteration count, elapsed Duration, and per-branch win counters, and exposes safely-derived rates so that empty runs can never inject NaN downstream.
The TradeoffSummary reducer is the deliverable most callers will reach for. It collapses two runs into integer percentages and two boolean starvation flags, which means downstream assertions can compare exact values with assert_eq! rather than chasing floating-point tolerance — and the test suite gains a single place to detect either policy silently flipping its behaviour.
Operationally, the take-away from the numbers is the rule of thumb the rest of the article has been pointing at: reach for biased; when one branch carries a real priority — shutdown, cancellation, timeouts — and where blocking the others for a poll cycle is a feature, not a bug. Stay with the fair default everywhere else, especially in long-running loops where every branch must eventually make progress, because deterministic priority and guaranteed liveness for low-priority branches are mutually exclusive properties.
This is also the last building block the article needs. The combination of "I understand cancellation safety" (step 3), "I have a cancel-survivable read primitive" (step 4), "I can pin poll order when it matters" (steps 5 and 6), and "I know the cost of each policy" (this step) is the full toolkit for writing tokio::select! bodies you actually want in production.
Repository
The state of the code after this step: b0d6fd3
Repository
Full source at https://github.com/vytharion/tokio-select-cancellation-safety-biased.
Walk the lessons by stepping through the git commits in the repo — each major step has its own commit you can git checkout and rerun.