Tokio Async Code Review

Tokio Async Code Review

Review Workflow

1. Check Cargo.toml — Note tokio feature flags (full, rt-multi-thread, macros, sync, etc.). Missing features cause confusing compile errors. 2. Check runtime setup — Is #[tokio::main] or manual runtime construction used? Multi-thread vs current-thread? 3. Scan for blocking — Search for std::fs, std::net, std::thread::sleep, CPU-heavy loops in async functions. 4. Check channel usage — Match channel type to communication pattern (mpsc, broadcast, oneshot, watch). 5. Check sync primitives — Verify correct mutex type, proper guard lifetimes, no deadlock potential.

Gates (objective passes before conclusions)

Complete in order for the review scope. Do not assert Critical or Major until the relevant gate passes.

1. Dependency surface — Read the crate (and workspace, if inherited) Cargo.toml that supplies tokio. Pass: Written note of tokio version and enabled features, or explicit statement that there is no direct tokio dependency and where it comes from (workspace/path). 2. Runtime model — Locate runtime construction (#[tokio::main], Runtime::builder, tests, or library with no owned runtime). Pass: One line naming flavor (multi_thread / current_thread / tests-only / none) and where it is defined. 3. Blocking inventory — Search reviewed paths for blocking APIs (std::fs::, std::net:: without async wrappers, std::thread::sleep, heavy CPU loops in async fn). Pass: Each hit listed as path:line (or tool output excerpt), or explicit “no blocking patterns found in reviewed async code” after the search. 4. Protocol — Load the review-verification-protocol skill. Pass: Its pass conditions met before any finding is reported (file:line evidence for asserted issues).

Output Format

Report findings as:

[FILE:LINE] ISSUE_TITLE
Severity: Critical | Major | Minor | Informational
Description of the issue and why it matters.

Quick Reference

Review Checklist

Runtime Configuration

• [ ] Tokio features in Cargo.toml match actual usage
• [ ] Runtime flavor matches workload (multi_thread for I/O-bound, current_thread for simpler cases)
• [ ] #[tokio::test] used for async tests (not manual runtime construction)
• [ ] Worker thread count configured appropriately for production

Task Management

• [ ] spawn return values (JoinHandle) are tracked, not silently dropped
• [ ] spawn_blocking used for CPU-heavy or synchronous I/O operations
• [ ] Tasks respect cancellation (via CancellationToken, select!, or shutdown channels)
• [ ] JoinError (task panic or cancellation) is handled, not just unwrapped
• [ ] tokio::select! branches are cancellation-safe
• [ ] Native async fn in traits used instead of async-trait crate where possible (stable since Rust 1.75)
• [ ] RPIT lifetime capture reviewed in async contexts — -> impl Future now captures all in-scope lifetimes in edition 2024

Sync Primitives

• [ ] tokio::sync::Mutex used when lock is held across .await; std::sync::Mutex for short non-async sections
• [ ] No mutex guard held across await points (deadlock risk)
• [ ] Semaphore used for limiting concurrent operations (not ad-hoc counters)
• [ ] RwLock used when read-heavy workload (many readers, infrequent writes)
• [ ] Notify used for simple signaling (not channel overhead)
• [ ] std::sync::LazyLock used instead of once_cell::sync::Lazy or lazy_static! for runtime-initialized singletons (stable since Rust 1.80)
• [ ] if let lock guard patterns reviewed for edition 2024 temporary scoping — temporaries drop earlier, may change borrow validity

Channels

• [ ] Channel type matches pattern: mpsc for back-pressure, broadcast for fan-out, oneshot for request-response, watch for latest-value
• [ ] Bounded channels have appropriate capacity (not too small = deadlock, not too large = memory)
• [ ] SendError / RecvError handled (indicates other side dropped)
• [ ] Broadcast Lagged errors handled (receiver fell behind)
• [ ] Channel senders dropped when done to signal completion to receivers

Timer and Sleep

• [ ] tokio::time::sleep used instead of std::thread::sleep
• [ ] tokio::time::timeout wraps operations that could hang
• [ ] tokio::time::interval used correctly (.tick().await for periodic work)

Severity Calibration

Critical

• Blocking I/O (std::fs::read, std::net::TcpStream) in async context without spawn_blocking
• Mutex guard held across .await point (deadlock potential)
• std::thread::sleep in async function (blocks runtime thread)
• Unbounded channel where back-pressure is needed (OOM risk)

Major

• JoinHandle silently dropped (lost errors, zombie tasks)
• Missing select! cancellation safety consideration
• Wrong mutex type (std vs tokio) for the use case
• Missing timeout on network/external operations

Minor

• tokio::spawn for trivially small async blocks (overhead > benefit)
• Overly large channel buffer without justification
• Manual runtime construction where #[tokio::main] suffices
• std::sync::Mutex where contention is high enough to benefit from tokio's async mutex

Informational

• Suggestions to use tokio-util utilities (e.g., CancellationToken)
• Tower middleware patterns for service composition
• Structured concurrency with JoinSet
• Migration from async-trait crate to native async fn in traits
• Migration from once_cell / lazy_static to std::sync::LazyLock
• Using #[expect(lint)] instead of #[allow(lint)] for self-cleaning suppression

Valid Patterns (Do NOT Flag)

• `std::sync::Mutex` for short critical sections — tokio docs recommend this when no .await is inside the lock
• `tokio::spawn` without explicit join — Valid for background tasks with proper shutdown signaling
• Unbuffered channel capacity of 1 — Valid for synchronization barriers
• `#[tokio::main(flavor = "current_thread")]` in simple binaries — Not every app needs multi-thread runtime
• `clone()` on `Arc<T>` before `spawn` — Required for moving into tasks, not unnecessary cloning
• Large broadcast channel capacity — Valid when lagged errors are expensive (event sourcing)
• Native `async fn` in traits without `async-trait` — Stable since 1.75; the crate is still valid for dyn dispatch cases
• `+ use<'a>` on `-> impl Future` returns — Correct edition 2024 precise capture syntax to limit lifetime capture
• `#[expect(clippy::type_complexity)]` on complex async types — Self-cleaning alternative to #[allow], warns when suppression is no longer needed

Before Submitting Findings

After Gates, apply the review-verification-protocol skill to every reported issue (evidence and dispositions per that skill).

Channels

Choosing the Right Channel

mpsc (Multi-Producer Single-Consumer)

The most common channel. Use bounded for back-pressure, unbounded only when you have external flow control.

// Bounded - preferred, provides back-pressure
let (tx, mut rx) = tokio::sync::mpsc::channel::<Event>(100);

// Unbounded - use with caution (OOM risk)
let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<Event>();

Capacity Sizing

• Too small (1-10): senders block frequently, throughput suffers
• Too large (100K+): memory pressure, defeats back-pressure purpose
• Rule of thumb: 2-4x the expected burst size

Graceful Shutdown via Drop

When all senders are dropped, rx.recv() returns None. This is the idiomatic way to signal "no more items."

// Producer side
drop(tx); // signals completion

// Consumer side
while let Some(item) = rx.recv().await {
    process(item);
}
// Loop exits when all senders dropped

Common Mistakes

// BAD - holding tx clone prevents shutdown
let tx_clone = tx.clone();
drop(tx);
// rx.recv() will never return None because tx_clone still exists

// BAD - send without handling closed channel
tx.send(item).await.unwrap(); // panics if receiver dropped

// GOOD - handle send errors
if tx.send(item).await.is_err() {
    tracing::warn!("receiver dropped, stopping producer");
    break;
}

broadcast

Every active subscriber receives every message. Messages are stored in a shared ring buffer.

let (tx, _rx) = tokio::sync::broadcast::channel::<Event>(16_384);

// Each subscriber gets their own receiver
let mut rx1 = tx.subscribe();
let mut rx2 = tx.subscribe();

Handling Lag

When a receiver falls behind, older messages are overwritten. The receiver gets RecvError::Lagged(n) indicating how many messages were missed.

loop {
    match rx.recv().await {
        Ok(event) => handle(event),
        Err(broadcast::error::RecvError::Lagged(n)) => {
            tracing::warn!(missed = n, "receiver lagged, some events lost");
            // Continue processing — data may be stale
        }
        Err(broadcast::error::RecvError::Closed) => break,
    }
}

Capacity Considerations

Broadcast stores messages until the slowest receiver consumes them (up to capacity). Size the buffer for the slowest expected consumer, not the average case.

oneshot

Single value, single use. The sender can only send once, and the receiver can only receive once.

let (tx, rx) = tokio::sync::oneshot::channel::<Result<Response, Error>>();

// Responder
tokio::spawn(async move {
    let result = compute().await;
    let _ = tx.send(result); // receiver may have been dropped
});

// Requester
match rx.await {
    Ok(result) => handle(result),
    Err(_) => tracing::error!("responder dropped without sending"),
}

Common Pattern: Request-Response

struct Request {
    data: InputData,
    reply: oneshot::Sender<Result<OutputData, Error>>,
}

// Client
let (tx, rx) = oneshot::channel();
request_tx.send(Request { data, reply: tx }).await?;
let response = rx.await??;

// Server
while let Some(req) = request_rx.recv().await {
    let result = process(req.data).await;
    let _ = req.reply.send(result);
}

watch

Holds the latest value. Receivers see only the most recent value, not a queue. Good for config changes or state snapshots.

let (tx, rx) = tokio::sync::watch::channel(Config::default());

// Update
tx.send(new_config)?;

// Read latest (non-blocking)
let current = rx.borrow().clone();

// Wait for changes
let mut rx = rx.clone();
loop {
    rx.changed().await?;
    let config = rx.borrow().clone();
    apply_config(config);
}

Dual-Channel Pattern (Event Bus)

For systems that need both real-time fan-out and durable persistence, combine broadcast (real-time) with mpsc (persistence):

struct EventBus {
    broadcast_tx: broadcast::Sender<Arc<Event>>,
    persist_tx: mpsc::Sender<Arc<Event>>,
}

impl EventBus {
    async fn emit(&self, event: Event) {
        let event = Arc::new(event);

        // Fan-out to all subscribers (best-effort)
        let _ = self.broadcast_tx.send(Arc::clone(&event));

        // Durable events go to persistence channel (back-pressure aware)
        if event.event_type.is_durable() {
            if let Err(e) = self.persist_tx.send(event).await {
                tracing::error!(error = %e, "persistence channel closed");
            }
        }
    }
}

Review Questions

1. Is the channel type matched to the communication pattern? 2. Are bounded channels sized appropriately for the workload? 3. Are SendError / RecvError handled (not unwrapped)? 4. Is broadcast Lagged error handled gracefully? 5. Are all sender clones dropped to allow clean shutdown? 6. Is watch used instead of broadcast for latest-value-only patterns?

Pinning, Cancellation, and Async Internals

Pin<P<T>> Semantics

Pin<P> wraps a pointer type P (e.g., &mut T, Box<T>) and guarantees the target T will not move after being pinned. Required for self-referential types like async state machines, where internal references would be invalidated by a move.

• Pin::new_unchecked() is unsafe -- caller must guarantee the referent won't move
• get_unchecked_mut() is unsafe -- caller must not move T through the returned &mut T
• Pin always implements Deref<Target = T> safely (shared refs can't move T)
• When required: Futures from async fn/blocks (self-referential across .await points), any struct storing data and pointers into that data

Unpin Trait

Unpin is an auto-trait indicating a type is safe to move out of a Pin. Most standard types are Unpin. Compiler-generated futures from async blocks are !Unpin.

// Unpin types can use the safe Pin::new constructor
let mut fut = ready(42);
let pinned = Pin::new(&mut fut); // safe, ready() is Unpin

// !Unpin types require heap or stack pinning
let fut = async { do_work().await };
let pinned = Box::pin(fut); // heap pinning, always safe

Stack vs Heap Pinning

`Box::pin(value)` (heap): Always safe. Allocates on the heap, so the Pin can move freely without moving T.

`std::pin::pin!(value)` (stack, stable since 1.68): Avoids heap allocation. Pins the value to the current stack frame via variable shadowing.

use std::pin::pin;

// GOOD - stack pinning with std::pin::pin! (stable 1.68+)
let fut = pin!(async { long_running().await });
fut.await;

// GOOD - heap pinning when you need to store or move the pinned future
let fut = Box::pin(async { long_running().await });
tokio::spawn(fut);

Flag when: pin_mut! from pin-utils or futures crate is used instead of std::pin::pin! on Rust 1.68+.

Future Trait Internals

The actual Future trait requires pinning and a waker context:

trait Future {
    type Output;
    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output>;
}

• `Poll::Pending`: Future cannot make progress. Must arrange for cx.waker().wake() to be called when progress is possible.
• `Poll::Ready(T)`: Future has resolved. Do not poll again (may panic).
• Waker contract: If poll returns Pending, the future must ensure wake() is called eventually. Leaf futures store the waker where the event source can trigger it.

Executor Model

Executors manage tasks (top-level futures) and decide which to poll when wakers fire.

• Work-stealing (tokio multi-thread): Multiple threads share a task queue. Idle threads steal work from busy ones. Good for I/O-bound workloads with many tasks.
• Single-threaded (tokio current_thread): One thread polls all tasks. No Send requirement on futures. Simpler, lower overhead, but no parallelism.

Check for: Blocking operations in async context. A future that runs >1ms without yielding Pending starves other tasks on the same executor thread.

Stream Trait (Async Iteration)

Stream (from futures crate) is the async equivalent of Iterator, with poll_next(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>>.

Use `Stream` when: Producing a sequence of values from a single source (file chunks, database rows, transformed events).

Use channels when: Multiple producers need to send to one consumer, or you need back-pressure across task boundaries.

Valid pattern: tokio_stream::StreamExt for combinators like .map(), .filter(), .throttle() on streams.

Cancellation Patterns

Drop as Cancellation

Dropping a future cancels it. This is fundamental to how tokio::select! works: unselected branches are dropped.

// Dropping the JoinHandle does NOT cancel the task (it detaches)
let handle = tokio::spawn(work());
drop(handle); // task keeps running!

// To cancel a spawned task, call abort()
let handle = tokio::spawn(work());
handle.abort(); // task is cancelled

CancellationToken + Graceful Shutdown

Hierarchical cancellation for coordinated shutdown. Child tokens cancel when parents do. Combine with JoinSet for clean drain.

use tokio_util::sync::CancellationToken;

let token = CancellationToken::new();
let mut set = JoinSet::new();

for worker in workers {
    let child = token.child_token();
    set.spawn(async move {
        tokio::select! {
            _ = child.cancelled() => worker.drain().await,
            _ = worker.run() => {}
        }
    });
}

// On SIGTERM: cancel parent, then drain all tasks
token.cancel();
while let Some(result) = set.join_next().await {
    result.expect("worker panicked");
}

select! Deep Semantics

• Branch priority: When multiple branches are ready simultaneously, tokio::select! picks one randomly by default. Use biased; to evaluate top-to-bottom.
• Cancellation safety: Unselected branches are dropped. A future is cancellation-safe if dropping it at any .await point doesn't lose data.

// RISKY - read_exact buffers internally, partial reads lost on cancel
tokio::select! {
    result = reader.read_exact(&mut buf) => { ... }
    _ = token.cancelled() => { return; }
}

// SAFER - use cancellation-safe recv()
tokio::select! {
    msg = rx.recv() => { ... }
    _ = token.cancelled() => { return; }
}

Flag when: read_exact, read_to_end, or custom buffering futures appear in select! branches without cancellation safety analysis.

Blocking Bridge

// spawn_blocking: preferred for most blocking work
let hash = tokio::task::spawn_blocking(move || {
    compute_hash(&data)
}).await?;

// block_in_place: only on multi-thread runtime, avoids thread pool queue
tokio::task::block_in_place(|| {
    std::fs::write(path, &data)?;
    Ok::<_, std::io::Error>(())
})?;

Flag when: block_in_place is used on current_thread runtime (it will panic).

Memory Ordering in Async Context

Async task scheduling provides happens-before relationships:

• tokio::spawn establishes happens-before between the spawning code and the spawned task's first poll
• .await on a JoinHandle establishes happens-before between the task's completion and the awaiting code
• Waker mechanics ensure proper ordering between wake() calls and subsequent poll()

When using atomics across async tasks, Ordering::Relaxed is usually sufficient for simple counters and flags, because task scheduling already provides synchronization. Use Acquire/Release only when you need ordering guarantees beyond what the runtime provides (e.g., custom lock-free structures shared across tasks).

Review Questions

1. Are !Unpin futures pinned correctly before polling (via Box::pin or std::pin::pin!)? 2. Are futures held across .await points reviewed for size (large futures = excessive memcpy)? 3. Is cancellation handled explicitly (CancellationToken, abort, or select) for long-running tasks? 4. Are select! branches cancellation-safe, or is data loss possible on cancel? 5. Is spawn_blocking used for blocking work instead of running it directly in async context? 6. Is block_in_place avoided on current_thread runtime? 7. Are dropped JoinHandles intentional (detached tasks) or accidental (lost errors)?

Sync Primitives

Choosing the Right Primitive

Mutex decision matrix: std vs tokio vs parking_lot

Default to std::sync::Mutex even in async code, provided the critical section is short and contains no .await. This is what the Tokio tutorial itself recommends.

tokio::sync::Mutex is a Semaphore under the hood. The Tokio docs and Mara both note it is materially more expensive (commonly cited as ~3x slower) than std::sync::Mutex for short critical sections because of its async-wakeup machinery. Reserve it for the case where you genuinely cannot extract or clone the data out before the .await.

parking_lot::Mutex::lock() returns the guard directly — no Result. A .unwrap() here is a smell. Its RwLock is fair by default; std::sync::RwLock is platform-dependent and can starve writers (or, on glibc, readers) — see the RwLock section below.

// GOOD - std::sync::Mutex in async code, no .await inside the critical section
use std::sync::Mutex;
struct Counter(Mutex<u64>);

impl Counter {
    async fn bump_and_log(&self) {
        let n = {
            let mut count = self.0.lock().unwrap();
            *count += 1;
            *count
        }; // guard dropped here, before any .await
        log(n).await;
    }
}

// GOOD - tokio::sync::Mutex only when the lock truly spans an await
use tokio::sync::Mutex;
struct Cache(Mutex<HashMap<String, Data>>);

impl Cache {
    async fn get_or_fetch(&self, key: &str) -> Data {
        let mut cache = self.0.lock().await;
        if let Some(data) = cache.get(key) { return data.clone(); }
        let data = fetch(key).await; // lock held across .await — intentional
        cache.insert(key.to_owned(), data.clone());
        data
    }
}

Review checks:

• [FILE:LINE] TOKIO_MUTEX_FOR_SHORT_SECTION — tokio::sync::Mutex wrapping data whose critical sections contain no .await. Replace with std::sync::Mutex.
• [FILE:LINE] PARKING_LOT_UNWRAP_ON_LOCK — parking_lot::Mutex::lock().unwrap(). parking_lot locks do not return Result; drop the .unwrap().

std::sync::MutexGuard held across .await (canonical async footgun)

This is not just a performance issue. It is a deadlock hazard.

The executor wants to suspend the task at the .await. The std::sync::MutexGuard is still alive, holding the OS lock. If any other task on the same multi-threaded runtime worker takes the same lock and blocks, the runtime worker is stuck — the holder cannot be polled because its future is parked, and the contender cannot make progress because it owns the worker thread. On a single-threaded runtime, the deadlock is immediate.

std::sync::MutexGuard is !Send on most platforms, so the compiler will reject the future being spawned onto a multi-threaded runtime. On LocalSet, current-thread runtimes, or with parking_lot's send_guard feature, it compiles silently — and is still wrong.

The clippy lint clippy::await_holding_lock catches the common cases. Treat it as an error in any async crate.

Dropping the guard explicitly before the .await (drop(guard); other.await;) is correct in principle, but the borrow-region the compiler tracks does not always end at the drop() call — block-scoping is more robust.

// BAD - guard is alive across the .await; deadlock-prone, !Send so won't spawn
async fn lookup(state: &Mutex<HashMap<String, String>>, key: &str) -> Option<String> {
    let guard = state.lock().unwrap();
    let value = guard.get(key);
    fetch_metadata(key).await; // guard still held here
    value.cloned()
}

// GOOD - block-scope drops the guard before any .await
async fn lookup(state: &Mutex<HashMap<String, String>>, key: &str) -> Option<String> {
    let value = {
        let guard = state.lock().unwrap();
        guard.get(key).cloned()
    }; // guard dropped here
    fetch_metadata(key).await;
    value
}

Review checks:

• [FILE:LINE] STD_MUTEX_HELD_ACROSS_AWAIT — std::sync::MutexGuard (or parking_lot::MutexGuard) is alive across an .await. Block-scope to drop before the await, or move to tokio::sync::Mutex if the lock must span it.
• [FILE:LINE] EXPLICIT_DROP_DOES_NOT_RELEASE_FOR_AWAIT — drop(guard); x.await; used to "release" the lock before an await. The compiler may keep the borrow region open. Use block-scoping instead.

Semaphore as a generalized lock and back-pressure primitive

tokio::sync::Semaphore is the right primitive for permit-based concurrency limiting: bounding parallel HTTP fetches, capping in-flight DB queries, throttling a fan-out, or implementing a connection pool. Unlike a Mutex, permits compose — you can hold N at once.

Two acquire flavours:

• acquire() returns a SemaphorePermit<'_> borrowed from &self. Cheaper, but tied to the lifetime — useless for tokio::spawn'd tasks.
• acquire_owned() consumes an Arc<Semaphore> and returns an OwnedSemaphorePermit: 'static. Required for spawning, more allocation.

use tokio::sync::Semaphore;
use std::sync::Arc;

// GOOD - bounded fan-out with explicit back-pressure
let sem = Arc::new(Semaphore::new(10));
let mut handles = Vec::new();
for item in items {
    let permit = sem.clone().acquire_owned().await.unwrap(); // back-pressure here
    handles.push(tokio::spawn(async move {
        let result = process(item).await;
        drop(permit); // released on task completion (or here, explicit)
        result
    }));
}

// BAD - unbounded spawn; producer outruns consumer, memory grows without limit
for item in items {
    tokio::spawn(async move { process(item).await });
}

For try-acquire (non-blocking):

match sem.try_acquire() {
    Ok(permit) => { /* proceed */ }
    Err(_) => { /* at capacity, back off or shed load */ }
}

Semaphore::close() permanently fails all future and pending acquires — useful for shutdown signalling without a separate channel.

Review checks:

• [FILE:LINE] UNBOUNDED_FAN_OUT_NO_SEMAPHORE — for ... { tokio::spawn(...) } over an unbounded input stream with no semaphore, channel, or buffer_unordered capping concurrency. Producer can outpace the runtime.
• [FILE:LINE] SEMAPHORE_REF_PERMIT_ACROSS_SPAWN — sem.acquire().await (returns borrowed SemaphorePermit<'_>) used in a closure passed to tokio::spawn. The lifetime cannot be 'static. Use acquire_owned() on an Arc<Semaphore>.

RwLock fairness and writer/reader starvation

Allows multiple concurrent readers or one exclusive writer.

use tokio::sync::RwLock;

let config = Arc::new(RwLock::new(Config::default()));

// Many readers concurrently
let cfg = config.read().await;
let port = cfg.port;
drop(cfg);

// Exclusive writer
let mut cfg = config.write().await;
cfg.port = 8080;

tokio::sync::RwLock prevents the classic writer-starvation case — a constant stream of readers cannot indefinitely block a queued writer, because once a writer is queued, new readers are forced to wait. But the reverse is now possible: readers can be delayed if writers are constantly queueing. For a write-heavy or write-bursty workload, this can convert an RwLock into a worse-than-Mutex serialization point. Benchmark before assuming RwLock wins.

The same write-preferring discipline causes recursive reads on the same task to deadlock: if your task is holding a read guard and a writer queues, asking for a second read guard from the same task waits for that writer, who is waiting for you.

parking_lot::RwLock is unfair by default (no writer-preference; readers may starve writers, writers may starve readers depending on contention) but has a fair mode (fair_unlock) and supports upgradable read guards. In sync code that needs predictable behavior, prefer parking_lot::RwLock over std::sync::RwLock (whose fairness is platform-dependent — glibc can starve readers, musl can starve writers).

Reach for `Mutex` when in doubt. With short critical sections, a Mutex is often faster than an RwLock even with many readers — the bookkeeping cost of RwLock exceeds the savings from reader concurrency until critical sections are long enough to amortize it. For read-mostly state that is occasionally replaced (rather than mutated in place), arc_swap::ArcSwap beats both — readers are wait-free.

Review checks:

• [FILE:LINE] TOKIO_RWLOCK_NESTED_READ — Two read().await calls taken in nested scopes on the same task. Write-preferring fairness will deadlock if a writer queues between them. Hoist into a single read scope, or refactor.
• [FILE:LINE] RWLOCK_FOR_SHORT_SECTION — RwLock wrapping data whose critical sections are a few field reads. Benchmark against Mutex; the reader-concurrency win likely does not amortize the overhead.
• [FILE:LINE] RWLOCK_ARC_FOR_READ_MOSTLY — RwLock<Arc<T>> (or RwLock<T> with Clone-then-mutate-then-replace) used for read-mostly, write-rare state. Use arc_swap::ArcSwap<T> for wait-free reads.

Atomics beat locks for single-word state

A single Mutex<u64> or Mutex<bool> for a counter or a flag is almost always the wrong primitive. Prefer the matching atomic (AtomicU64, AtomicBool, AtomicUsize, AtomicPtr):

• No lock contention.
• No MutexGuard to accidentally hold across .await.
• Smaller — an atomic is the size of its underlying type; a Mutex carries the lock word plus the OS handle.
• Compatible with Relaxed ordering for the common cases (statistics, stop flags, monotonic counters) — see [../../rust-code-review/references/memory-ordering.md] for ordering choice and [../../rust-code-review/references/lock-free-patterns.md] for CAS-loop patterns.

// BAD - Mutex for a counter; lock contention, awkward in async
use std::sync::{Arc, Mutex};
let count = Arc::new(Mutex::new(0u64));
for _ in 0..n {
    let c = count.clone();
    tokio::spawn(async move { *c.lock().unwrap() += 1; });
}

// GOOD - atomic; no lock, no guard, no await question
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering::Relaxed};
let count = Arc::new(AtomicU64::new(0));
for _ in 0..n {
    let c = count.clone();
    tokio::spawn(async move { c.fetch_add(1, Relaxed); });
}

The mutex form is also wrong if you ever read multiple counters together and treat them as a snapshot — separate atomic loads can interleave with another thread's increments. For coherent multi-field snapshots you do need a Mutex (or pack the fields into one atomic word with bit-fields). See [../../rust-code-review/references/concurrency-primitives.md] for that case.

Review checks:

• [FILE:LINE] ARC_MUTEX_FOR_COPY_TYPE — Arc<Mutex<T>> where T is Copy and primitive-sized (u8..u64, bool, usize, raw pointer) used for a single piece of state. Replace with Arc<Atomic*>. Document the chosen Ordering.
• [FILE:LINE] ATOMIC_SNAPSHOT_TORN — Multiple Atomic*::load calls treated as a coherent snapshot by the caller. Either use a Mutex over the struct, or pack the fields into one atomic word.

tokio::sync::Notify and the Notified future

Notify is the cheapest "wake one waiter" primitive in Tokio. Reach for it when you have a clear state-change event but no value to transfer (a channel would be the wrong tool) and no shared count to track (a Semaphore would be overkill).

The lost-wakeup hazard: Notify permits coalesce — if two notifications arrive before any task calls notified().await, only one wakeup remains. You must create the `Notified` future before re-checking the state; otherwise a notification arriving between the state check and the .notified().await will appear as no notification at all.

// BAD - lost wakeup: notify can fire between the check and the await
async fn wait_ready(notify: &Notify, ready: &AtomicBool) {
    while !ready.load(Acquire) {
        notify.notified().await; // notification before this line is lost
    }
}

// GOOD - register the future first, then re-check, then await
async fn wait_ready(notify: &Notify, ready: &AtomicBool) {
    loop {
        let notified = notify.notified();         // register first
        if ready.load(Acquire) { return; }        // then check
        notified.await;                           // then sleep
    }
}

Use a channel (mpsc, watch, broadcast) when you need to transfer a value. Use Notify only when the wakeup itself is the signal and there is associated state (an AtomicBool, a queue length, etc.) the waiter can re-check after waking.

Review checks:

• [FILE:LINE] NOTIFY_REGISTER_AFTER_CHECK — if !ready { notify.notified().await } or while !ready { notify.notified().await } without first constructing the Notified future before the state check. Lost-wakeup hazard.
• [FILE:LINE] NOTIFY_WITHOUT_STATE — Notify::notify_one() used as a wake mechanism with no associated ready/queue-length state for the waiter to re-check. Notify coalesces; pair it with an AtomicBool / watch channel, or use a channel directly.

OnceCell, OnceLock, and LazyLock in async code

std::sync::LazyLock is effectively poisoned by an init panic: the panic propagates, every subsequent access also panics, and there is no clear_poison analogue. For long-running services with fallible initialization, prefer OnceLock::get_or_try_init and store a Result<T, E> (or store a sentinel and retry) so the program survives a transient failure during cold start.

OnceLock::wait() (stable 1.86) blocks the current thread until the value is set — use it instead of busy-polling get().

tokio::sync::OnceCell::get_or_init deadlocks if the initializing closure re-enters the same cell — same hazard as the std version. Precompute or restructure.

// BAD - LazyLock with fallible init; one bad config file kills the whole service
static CFG: LazyLock<Config> = LazyLock::new(|| Config::load().unwrap());

// GOOD - OnceLock with try-init; caller decides how to handle the error
static CFG: OnceLock<Config> = OnceLock::new();
fn cfg() -> Result<&'static Config, ConfigError> {
    CFG.get_or_try_init(Config::load)
}

// GOOD - tokio::sync::OnceCell for async initialization
static POOL: OnceCell<Pool> = OnceCell::const_new();
async fn pool() -> &'static Pool {
    POOL.get_or_init(|| async { Pool::connect(&DB_URL).await.unwrap() }).await
}

For single-threaded or non-Sync contexts, use std::cell::LazyCell instead.

Migration: replace once_cell::sync::Lazy and lazy_static! in any crate targeting MSRV ≥ 1.80 with std::sync::LazyLock.

Review checks:

• [FILE:LINE] LAZYLOCK_FALLIBLE_INIT — LazyLock::new whose closure can panic on bad input (missing env var, malformed config). Panic is unrecoverable; cell stays poisoned for the process lifetime. Use OnceLock::get_or_try_init instead.

if let Temporary Scope Changes (Edition 2024)

In Rust 2024, temporaries in if let conditions are dropped at the end of the if let condition, not at the end of the block. This affects async lock guard patterns.

// Edition 2021 - guard lives through the if-let body
if let Some(val) = state.lock().await.get("key") {
    // guard is still alive here in edition 2021
    do_work(val).await; // holding the lock across await — risky but compiles
}

// Edition 2024 - guard is dropped after the condition evaluates
// val would be a dangling reference — this may fail to compile
if let Some(val) = state.lock().await.get("key") {
    do_work(val).await; // guard already dropped!
}

// GOOD - explicit binding extends the guard's lifetime
let guard = state.lock().await;
if let Some(val) = guard.get("key") {
    do_work(val).await;
}
drop(guard);

// GOOD - clone the value to avoid depending on guard lifetime
if let Some(val) = state.lock().await.get("key").cloned() {
    do_work(val).await; // val is owned, guard already dropped — safe
}

This also applies to while let and match with temporary-producing expressions. Review any pattern where a lock guard is created inline in a conditional.

Common Mistakes

Deadlock via Lock Ordering

// BAD - potential deadlock if another task locks B then A
let _a = state_a.lock().await;
let _b = state_b.lock().await;

// GOOD - always lock in consistent order, or use a single lock

Forgetting to Drop Guards

// BAD - guard lives until end of scope, holding lock during await
let guard = state.lock().await;
let value = guard.get_value();
do_async_work(value).await; // guard still held!

// GOOD - extract value and drop guard
let value = {
    let guard = state.lock().await;
    guard.get_value().clone()
};
do_async_work(value).await;

Review Questions

1. Is the right sync primitive chosen for the access pattern? 2. Are mutex guards dropped before .await points? 3. Is lock ordering consistent to prevent deadlocks? 4. Is Semaphore used instead of ad-hoc concurrency limits? 5. Are std::sync vs tokio::sync primitives matched to their context? 6. Are once_cell / lazy_static usages replaced with std::sync::LazyLock where possible? 7. Do if let / while let patterns with inline lock guards account for edition 2024 temporary scoping?

Task Management

Spawning Tasks

tokio::spawn

Creates an independent task on the runtime. The spawned future must be Send + 'static.

// Basic spawn with error handling
let handle = tokio::spawn(async move {
    process(data).await
});

match handle.await {
    Ok(Ok(result)) => tracing::info!(?result, "task completed"),
    Ok(Err(e)) => tracing::error!(error = %e, "task failed"),
    Err(e) => tracing::error!(error = %e, "task panicked"),
}

tokio::spawn_blocking

Runs a closure on a dedicated thread pool for blocking operations. Returns a JoinHandle like spawn.

// CPU-heavy work belongs on blocking threads
let hash = tokio::task::spawn_blocking(move || {
    argon2::hash_password(&password, &salt)
}).await??;

// Synchronous file I/O
let contents = tokio::task::spawn_blocking(move || {
    std::fs::read_to_string(path)
}).await??;

JoinSet for Structured Concurrency

JoinSet manages a group of tasks with collective lifecycle control. Preferred over tracking individual JoinHandles when spawning dynamic numbers of tasks.

use tokio::task::JoinSet;

let mut set = JoinSet::new();
for item in items {
    set.spawn(async move {
        process(item).await
    });
}

// Collect all results
while let Some(result) = set.join_next().await {
    match result {
        Ok(Ok(value)) => results.push(value),
        Ok(Err(e)) => tracing::warn!(error = %e, "task failed"),
        Err(e) => tracing::error!(error = %e, "task panicked"),
    }
}

When a JoinSet is dropped, all tasks in it are cancelled (aborted). This provides automatic cleanup.

Cancellation

CancellationToken (tokio-util)

Hierarchical cancellation for structured shutdown. Child tokens are cancelled when parents are.

use tokio_util::sync::CancellationToken;

let token = CancellationToken::new();

// Worker respects cancellation
let child = token.child_token();
tokio::spawn(async move {
    loop {
        tokio::select! {
            _ = child.cancelled() => break,
            item = rx.recv() => {
                if let Some(item) = item {
                    process(item).await;
                }
            }
        }
    }
});

// On shutdown:
token.cancel(); // cancels all children

select! Cancellation Safety

When tokio::select! resolves one branch, other branches are dropped. A future is cancellation-safe if dropping it at any .await point doesn't lose data.

Cancellation-safe operations:

• tokio::sync::mpsc::Receiver::recv()
• tokio::sync::oneshot::Receiver::recv()
• tokio::time::sleep()
• tokio::io::AsyncReadExt::read() (data goes to caller's buffer)

NOT cancellation-safe:

• tokio::io::AsyncReadExt::read_exact() — partial reads are lost
• Custom futures that do internal buffering

// RISKY - read_exact may partially fill buffer then get cancelled
tokio::select! {
    result = reader.read_exact(&mut buf) => { ... }
    _ = cancel.cancelled() => { return; }
}

// SAFER - use read() and handle partial reads manually
tokio::select! {
    result = reader.read(&mut buf) => { ... }
    _ = cancel.cancelled() => { return; }
}

async fn in Traits (Rust 2024 Edition)

Since Rust 1.75, async fn works directly in trait definitions without the async-trait crate. This matters for tokio-based service patterns.

// BAD (edition 2024) - unnecessary async-trait dependency
#[async_trait::async_trait]
trait Handler: Send + Sync {
    async fn handle(&self, request: Request) -> Response;
}

// GOOD (edition 2024) - native async fn in traits
trait Handler: Send + Sync {
    fn handle(&self, request: Request) -> impl Future<Output = Response> + Send;
}

// GOOD (edition 2024) - also valid with async fn directly
trait Handler: Send + Sync {
    async fn handle(&self, request: Request) -> Response;
}

When `async-trait` is still needed:

• Trait objects (dyn Handler) — native async traits are not yet object-safe
• When you need Box<dyn Future> return types for dynamic dispatch

RPIT Lifetime Capture in Async Contexts

In edition 2024, -> impl Trait captures ALL in-scope lifetimes by default (including elided ones). This can cause unexpected borrow-checker errors in async code that returns impl Future.

// Edition 2021 - only captures 'a explicitly
fn process(data: &str) -> impl Future<Output = ()> {
    async { /* ... */ }
}

// Edition 2024 - now captures the lifetime of `data` by default
// This may cause "borrowed value does not live long enough" errors
fn process(data: &str) -> impl Future<Output = ()> {
    async { /* ... */ }
}

// GOOD - use precise capturing to opt out of capturing the borrow
fn process(data: &str) -> impl Future<Output = ()> + use<> {
    let owned = data.to_owned();
    async move { /* use owned */ }
}

// GOOD - if the future genuinely needs the borrow, capture it explicitly
fn process<'a>(data: &'a str) -> impl Future<Output = ()> + use<'a> {
    async { println!("{data}"); }
}

This is especially relevant for spawned tasks, which require 'static:

// BAD (edition 2024) - impl Future captures the borrow, can't spawn
fn make_task(config: &Config) -> impl Future<Output = ()> {
    let value = config.get_value();
    async move { use_value(value).await; }
}

// GOOD - precise capture excludes the borrow
fn make_task(config: &Config) -> impl Future<Output = ()> + use<> {
    let value = config.get_value();
    async move { use_value(value).await; }
}

Review Questions

1. Are all JoinHandles either awaited, stored, or deliberately dropped with comment? 2. Is spawn_blocking used for CPU-heavy or synchronous I/O work? 3. Are task groups managed with JoinSet instead of manual handle tracking? 4. Is cancellation implemented via CancellationToken or equivalent? 5. Are select! branches cancellation-safe? 6. Is async-trait crate used where native async fn in traits would suffice? 7. Do -> impl Future returns have correct lifetime capture behavior for edition 2024?