--- name: tokio-safety-auditor description: Industrial static AST and heuristic auditor for asynchronous Rust codebases using Tokio. Detects blocking calls in async runtime, Mutex misuse across await boundaries, tokio::select! cancellation safety hazards, and unmanaged thread pool starvation. --- # tokio-safety-auditor: Rust Asynchronous Safety & Runtime Integrity Guard This skill hardens AI agents (Claude Code, Cursor, Antigravity) writing asynchronous Rust using Tokio. General-purpose language models frequently generate async code that compiles successfully but causes threadpool starvation, deadlocks, or silent data corruption at runtime. ## Core Audit Domains ### 1. Synchronous Blocking I/O inside Async Tasks - **Hazard**: Calling synchronous operations inside `async fn` or `tokio::spawn` blocks the Tokio worker thread. Because Tokio runs an M:N cooperative scheduler with a small thread pool (typically equal to CPU core count), blocking one thread degrades throughput across all co-scheduled tasks. - **Forbidden Calls**: - `std::fs::File`, `std::fs::read`, `std::fs::write`, `std::fs::read_to_string` - `std::thread::sleep` - `std::net::TcpStream`, `std::net::UdpSocket` - Blocking database drivers (e.g. standard `rusqlite::Connection` without blocking isolation) - Synchronous HTTP clients (e.g. `reqwest::blocking::*`) - Long CPU-bound loops (> 100 microseconds without yielding) - **Required Remediation**: - Replace filesystem calls with `tokio::fs::*`. - Replace thread sleeps with `tokio::time::sleep`. - Offload unavoidable blocking operations or CPU-intensive serialization/decoding to `tokio::task::spawn_blocking`. - In tight computational loops, insert `tokio::task::yield_now().await` to prevent starving co-scheduled futures. ### 2. Mutex Pitfalls Across Await Points - **Hazard A (Deadlock & Inverted Priority)**: Holding a `std::sync::MutexGuard` across an `.await` boundary prevents other worker threads from acquiring the lock. If the awaiting task is rescheduled onto another thread or paused waiting on a dependency held by a blocked thread, the runtime deadlocks. - **Hazard B (Compile Failure / Send Trait Violation)**: `std::sync::MutexGuard` does not implement `Send` on many platforms, making the enclosing future `!Send`, preventing it from being passed to `tokio::spawn`. - **Decision Matrix**: - **Short, in-memory updates without await points**: Use `parking_lot::Mutex` or `std::sync::Mutex`, explicitly drop the guard before calling `.await` (use scoped `{}` blocks). - **Locks held across await points**: Use `tokio::sync::Mutex` or `tokio::sync::RwLock`. - **High-throughput state sharing**: Prefer message passing via `tokio::sync::mpsc` or lock-free atomics (`crossbeam::atomic`, `portable_atomic`) instead of coarse-grained async mutexes. ### 3. tokio::select! Cancellation Safety Hazards - **Hazard**: When a branch in `tokio::select!` completes, all other pending branches are immediately dropped. If a future holds partial state that cannot be cleanly resumed or aborted, dropping it leads to data loss or broken protocol framing. - **Cancellation-Unsafe Operations**: - `tokio::io::AsyncReadExt::read`: If dropped after reading partial bytes into a stack buffer, those bytes are permanently discarded. - `tokio::io::AsyncWriteExt::write_all`: Dropping midway leaves the stream partially written with no recovery mechanism. - Channels without persistent state: Calling `rx.recv()` in a branch without saving the received message before the next await point. - **Remediation**: - For continuous network streams, use framed protocols (`tokio_util::codec::FramedRead`, `FramedWrite`). - Pin the future outside the loop: `tokio::pin!(stream_fut);` and select on `&mut stream_fut`. - Ensure futures wrapped in `tokio::select!` are documented as cancellation-safe in the Tokio specification. ### 4. Thread Pool Starvation & Task Lifecycle - **Hazard**: Unbounded `tokio::task::spawn_blocking` can exhaust the Tokio blocking thread pool (default max 512 threads), causing extreme latency spikes and thread context switching overhead. - **Remediation**: - Bound concurrent blocking jobs using `tokio::sync::Semaphore`. - Always retain and manage `JoinHandle` values; handle panics explicitly using `join_handle.await.map_err(...)`. - Wire graceful shutdown using `tokio_util::sync::CancellationToken` or `tokio::sync::broadcast`. --- ## Agent System Prompt Directives When writing or refactoring asynchronous Rust code, you must enforce the following rules: ```markdown 1. Never call std::fs or std::thread::sleep inside async functions. Use tokio::fs and tokio::time. 2. For compute-heavy functions (>100µs) or blocking C-bindings, wrap them inside tokio::task::spawn_blocking. 3. Never hold a std::sync::MutexGuard across an .await point. If lock duration is short, wrap it in a local block { let guard = mutex.lock(); ... } before the await. If the lock must span across an await, use tokio::sync::Mutex. 4. When using tokio::select!, verify every branch future for cancellation safety. If using stream reading, employ tokio_util::codec or pin the future outside the loop. 5. Provide a Clippy configuration that flags clippy::await_holding_lock and disallowed synchronous blocking calls. ``` --- ## Concrete Bad vs Good Implementation Patterns ### Pattern 1: Filesystem Access ```rust // [INCORRECT: Stalls Tokio Worker Thread] pub async fn load_config_bad(path: &Path) -> Result { // std::fs blocks the executor thread let content = std::fs::read_to_string(path)?; let config: Config = serde_json::from_str(&content)?; Ok(config) } // [CORRECT: Native Async I/O] pub async fn load_config_good(path: &Path) -> Result { let content = tokio::fs::read_to_string(path).await?; let config: Config = serde_json::from_str(&content)?; Ok(config) } // [CORRECT: Heavy Parsing Offloaded to Blocking Pool] pub async fn load_and_parse_heavy(path: PathBuf) -> Result { let bytes = tokio::fs::read(path).await?; tokio::task::spawn_blocking(move || { let dataset: HugeDataset = bincode::deserialize(&bytes)?; Ok(dataset) }) .await .map_err(|e| AppError::TaskJoinError(e.to_string()))? } ``` ### Pattern 2: Mutex Usage Across Await Boundaries ```rust // [INCORRECT: std::sync::MutexGuard held across await] use std::sync::Mutex; pub async fn update_metrics_bad(metrics: &Mutex, client: &HttpClient) { let mut guard = metrics.lock().unwrap(); let stats = client.fetch_stats().await; // AWAIT POINT: Guard held, worker thread locked! guard.total_requests += stats.requests; } // [CORRECT OPTION A: Scope Guard Lifetime Before Await] pub async fn update_metrics_scoped(metrics: &std::sync::Mutex, client: &HttpClient) { let stats = client.fetch_stats().await; { let mut guard = metrics.lock().unwrap(); guard.total_requests += stats.requests; } // Guard dropped immediately after memory update } // [CORRECT OPTION B: Tokio Async Mutex When Lock Must Span Await] pub async fn acquire_token_good(token_mgr: &tokio::sync::Mutex) -> Result { let mut guard = token_mgr.lock().await; guard.refresh_token_if_expired().await?; Ok(guard.current_token().to_string()) } ``` ### Pattern 3: tokio::select! Cancellation Safety ```rust // [INCORRECT: AsyncReadExt::read inside select drops unconsumed buffer] use tokio::io::AsyncReadExt; pub async fn read_network_bad( mut reader: R, mut shutdown: tokio::sync::watch::Receiver, ) { let mut buf = [0u8; 1024]; loop { tokio::select! { // If shutdown fires while reader.read is pending, any internal partially read chunk is lost n = reader.read(&mut buf) => { match n { Ok(0) => break, Ok(bytes) => process_packet(&buf[..bytes]), Err(e) => eprintln!("Read error: {e}"), } } _ = shutdown.changed() => { println!("Graceful shutdown requested"); break; } } } } // [CORRECT: Use FramedRead with stateful Decoder or Pinned Stream] use tokio_util::codec::{BytesCodec, FramedRead}; use futures::StreamExt; pub async fn read_network_good( reader: R, mut shutdown: tokio::sync::watch::Receiver, ) { let mut framed = FramedRead::new(reader, BytesCodec::new()); loop { tokio::select! { biased; // Deterministic branch evaluation order _ = shutdown.changed() => { println!("Shutdown signal triggered"); break; } frame = framed.next() => { match frame { Some(Ok(bytes)) => process_packet(&bytes), Some(Err(err)) => eprintln!("Protocol error: {err}"), None => break, // Stream reached EOF } } } } } ``` --- ## Toolchain & Clippy Configuration Place this `clippy.toml` in your project root to enforce lint checks at the compiler level: ```toml # clippy.toml disallowed-methods = [ { path = "std::fs::read", reason = "Use tokio::fs::read inside async context" }, { path = "std::fs::read_to_string", reason = "Use tokio::fs::read_to_string inside async context" }, { path = "std::fs::write", reason = "Use tokio::fs::write inside async context" }, { path = "std::thread::sleep", reason = "Use tokio::time::sleep instead of blocking worker threads" }, ] ``` Add these lints to your `Cargo.toml`: ```toml [lints.clippy] await_holding_lock = "deny" await_holding_refcell_ref = "deny" disallowed_methods = "deny" ``` --- ## Automated Heuristic Scanner This skill includes an automated Python scanner (`scripts/audit_tokio.py`) that can be executed directly by the agent or pre-commit hooks: ```bash python3 scripts/audit_tokio.py --target ./src ``` It inspects Rust source files for: 1. `std::fs` and `std::thread::sleep` mentions inside `async fn` blocks. 2. `std::sync::Mutex` paired with `.await`. 3. High-risk `AsyncReadExt` patterns inside `tokio::select!`. 4. Bare `tokio::spawn` calls ignoring task error propagation. --- ## Pre-Merge Quality Gate Checklist - [ ] All `std::fs` operations in async paths replaced with `tokio::fs` or wrapped in `spawn_blocking`. - [ ] No `std::sync::MutexGuard` lives across an `.await` call (checked by `cargo clippy -- -D clippy::await_holding_lock`). - [ ] Any `tokio::select!` branches evaluated for cancellation safety; streaming I/O uses framed streams or pinned handles. - [ ] CPU-heavy computations exceeding 100 microseconds are delegated to `tokio::task::spawn_blocking`. - [ ] Graceful termination configured with `CancellationToken` or `broadcast::channel`.