rust-async-patterns

name	rust-async-patterns
description	Async/await patterns and tokio best practices for Rust. Use when implementing async functions, handling blocking operations, managing concurrent tasks, or working with async runtimes. Enforces proper resource budgets and backpressure handling. Use when this capability is needed.
metadata	{"author":"glubus"}

Rust Async Patterns Skill

This skill provides guidance on writing correct, efficient, and safe asynchronous Rust code using tokio and async/await.

When to use this skill

• Implementing async functions or services
• Converting blocking code to async
• Managing concurrent operations
• Working with tokio runtime
• Handling I/O-bound or CPU-bound work in async context
• Implementing timeouts and cancellation

Core Principles

1. Never Block the Async Runtime

Rule: CPU-intensive or blocking operations must run in spawn_blocking, never directly in async functions.

Why: Blocking the async executor starves other tasks and degrades performance.

Bad:

// BAD: Blocks the async runtime
pub async fn process_image(data: Vec<u8>) -> Result<Vec<u8>, Error> {
    // Heavy CPU work on async thread!
    let img = image::load_from_memory(&data)?;
    let resized = img.resize(100, 100, FilterType::Lanczos3);
    Ok(resized.to_bytes())
}

Good:

// GOOD: CPU work in blocking thread pool
pub async fn process_image(data: Vec<u8>) -> Result<Vec<u8>, Error> {
    tokio::task::spawn_blocking(move || {
        let img = image::load_from_memory(&data)?;
        let resized = img.resize(100, 100, FilterType::Lanczos3);
        Ok(resized.to_bytes())
    })
    .await
    .map_err(|e| Error::JoinError(e))?
}

2. Always Set Timeouts on I/O Operations

Rule: All network I/O, file I/O, and external calls must have explicit timeouts.

Why: Prevents indefinite hangs and resource exhaustion.

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

pub async fn fetch_data(url: &str) -> Result<String, Error> {
    let request = reqwest::get(url);
    
    match timeout(Duration::from_secs(10), request).await {
        Ok(Ok(response)) => Ok(response.text().await?),
        Ok(Err(e)) => Err(Error::RequestFailed(e)),
        Err(_) => Err(Error::Timeout),
    }
}

3. Bound Concurrent Operations

Rule: Never spawn unbounded concurrent tasks. Use semaphores or buffered channels.

Why: Prevents resource exhaustion and OOM conditions.

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

const MAX_CONCURRENT: usize = 10;

pub async fn process_batch(items: Vec<Item>) -> Result<Vec<Output>, Error> {
    let semaphore = Arc::new(Semaphore::new(MAX_CONCURRENT));
    let mut tasks = Vec::new();
    
    for item in items {
        let permit = semaphore.clone().acquire_owned().await?;
        
        let task = tokio::spawn(async move {
            let result = process_item(item).await;
            drop(permit); // Release permit
            result
        });
        
        tasks.push(task);
    }
    
    // Collect results
    let mut results = Vec::new();
    for   tasks {
        results.(task.??);
    }
    
    (results)
}

4. Use Structured Concurrency

Rule: Prefer tokio::select!, join!, and try_join! over manual task spawning when possible.

Why: Ensures proper cancellation and resource cleanup.

use tokio::try_join;

pub async fn load_chart_with_audio(
    chart_path: &Path,
    audio_path: &Path,
) -> Result<(RoxChart, AudioData), Error> {
    // Both load concurrently, both must succeed
    let (chart, audio) = try_join!(
        load_chart(chart_path),
        load_audio(audio_path)
    )?;
    
    Ok((chart, audio))
}

Common Async Patterns

Pattern 1: Async Function with Timeout

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

pub async fn decode_with_timeout(
    data: &[u8],
    max_duration: Duration,
) -> Result<RoxChart, CodecError> {
    timeout(max_duration, async {
        // Actual decode work
        decode_chart(data)
    })
    .await
    .map_err(|_| CodecError::Timeout)?
}

Pattern 2: CPU-Bound Work in Async Context

pub async fn compute_hash(data: Vec<u8>) -> Result<Hash, Error> {
    tokio::task::spawn_blocking(move || {
        // CPU-intensive hashing
        blake3::hash(&data)
    })
    .await
    .map_err(|e| Error::TaskFailed(e.to_string()))
}

Pattern 3: Concurrent Stream Processing

use futures::stream::{self, StreamExt};

pub async fn process_charts_concurrent(
    paths: Vec<PathBuf>,
    concurrency: usize,
) -> Vec<Result<RoxChart, CodecError>> {
    stream::iter(paths)
        .map(|path| async move {
            decode_chart(&path).await
        })
        .buffer_unordered(concurrency)
        .collect()
        .await
}

Pattern 4: Graceful Shutdown with Cancellation

use tokio::sync::watch;
use tokio::select;

pub async fn worker_with_shutdown(
    mut shutdown_rx: watch::Receiver<bool>,
) -> Result<(), Error> {
    loop {
        select! {
            _ = shutdown_rx.changed() => {
                info!("Shutdown signal received");
                break;
            }
            result = do_work() => {
                result?;
            }
        }
    }
    
    Ok(())
}

Pattern 5: Retry with Exponential Backoff

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

pub async fn retry_with_backoff<F, Fut, T, E>(
    mut operation: F,
    max_retries: u32,
) -> Result<T, E>
where
    F: FnMut() -> Fut,
    Fut: std::future::Future<Output = Result<T, E>>,
{
    let mut attempt = 0;
    
    loop {
        match operation().await {
            Ok(result) => return Ok(result),
            Err(e) if attempt >= max_retries => return Err(e),
            Err(_) => {
                attempt += 1;
                let delay = Duration::from_millis(100 * 2_u64.pow(attempt));
                sleep(delay).await;
            }
        }
    }
}

Decision Tree

When to use async vs sync?

1. Is this I/O-bound work? (network, file, database)

   • Yes → Use async
   • No → Continue to next question

2. Is this CPU-bound work?

   • Yes → Use spawn_blocking if called from async context
   • No → Continue to next question

3. Do I need concurrency?

   • Yes → Use async
   • No → Sync is simpler

When to use spawn vs spawn_blocking?

1. Is the work CPU-intensive? (image processing, hashing, compression)

   • Yes → Use spawn_blocking
   • No → Continue

2. Does it call blocking APIs? (std::fs, blocking network)

   • Yes → Use spawn_blocking
   • No → Use spawn for async work

How to handle multiple concurrent operations?

1. Must all succeed?

   • Yes → Use try_join!
   • No → Continue

2. Do I need first result?

   • Yes → Use select!
   • No → Use join! or manual task collection

Resource Budgets

Always define explicit limits:

// Configuration for async operations
pub struct AsyncConfig {
    /// Maximum concurrent blocking tasks
    pub max_blocking_tasks: usize,
    
    /// Maximum concurrent I/O operations
    pub max_io_concurrent: usize,
    
    /// Default timeout for I/O operations
    pub io_timeout: Duration,
    
    /// Maximum retry attempts
    pub max_retries: u32,
}

impl Default for AsyncConfig {
    fn default() -> Self {
        Self {
            max_blocking_tasks: num_cpus::get(),
            max_io_concurrent: 100,
            io_timeout: Duration::from_secs(30),
            max_retries: 3,
        }
    }
}

Common Mistakes to Avoid

❌ Blocking in async functions

// BAD
pub async fn process(data: Vec<u8>) -> Result<Vec<u8>, Error> {
    std::thread::sleep(Duration::from_secs(1)); // Blocks executor!
    Ok(data)
}

// GOOD
pub async fn process(data: Vec<u8>) -> Result<Vec<u8>, Error> {
    tokio::time::sleep(Duration::from_secs(1)).await;
    Ok(data)
}

❌ No timeout on I/O

// BAD
pub async fn fetch(url: &str) -> Result<String, Error> {
    Ok(reqwest::get(url).await?.text().await?)
}

// GOOD
pub async fn fetch(url: &str) -> Result<String, Error> {
    let response = timeout(
        Duration::from_secs(10),
        reqwest::get(url)
    ).await??;
    
    Ok(response.text().await?)
}

❌ Unbounded task spawning

// BAD
pub async fn process_all(items: Vec<Item>) -> Vec<Output> {
    let mut tasks = Vec::new();
    for item in items {
        tasks.push(tokio::spawn(process(item))); // Could spawn millions!
    }
    // ...
}

// GOOD
pub async fn process_all(items: Vec<Item>) -> Vec<Output> {
    stream::iter(items)
        .map(|item| process(item))
        .buffer_unordered(10) // Bounded concurrency
        .collect()
        .await
}

❌ Ignoring cancellation

// BAD
pub async fn long_running_task() {
    loop {
        do_work().await;
        // No way to cancel!
    }
}

// GOOD
pub async fn long_running_task(mut shutdown: watch::Receiver<bool>) {
    loop {
        select! {
            _ = shutdown.changed() => break,
            _ = do_work() => {}
        }
    }
}

Testing Async Code

#[cfg(test)]
mod tests {
    use super::*;
    use tokio::time::{pause, advance, Duration};

    #[tokio::test]
    async fn test_timeout_works() {
        let result = timeout(
            Duration::from_millis(100),
            async {
                tokio::time::sleep(Duration::from_secs(10)).await;
                "done"
            }
        ).await;
        
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn test_concurrent_processing() {
        let items = vec![1, 2, 3, 4, 5];
        let results = process_batch(items).await.unwrap();
        assert_eq!(results.len(), 5);
    }
}

Observability

Instrument async functions for tracing:

use tracing::{instrument, info, warn};

#[instrument(skip(data))]
pub async fn process_chart(
    chart_id: &str,
    data: Vec<u8>,
) -> Result<RoxChart, Error> {
    info!("Starting chart processing");
    
    let chart = tokio::task::spawn_blocking(move || {
        decode_chart(&data)
    })
    .await
    .map_err(|e| {
        warn!(error = %e, "Blocking task failed");
        Error::DecodeFailed
    })??;
    
    info!(note_count = chart.notes.len(), "Chart processed");
    Ok(chart)
}

Checklist

When implementing async code:

• CPU-bound work uses spawn_blocking
• All I/O operations have timeouts
• Concurrent operations are bounded
• Proper error handling with ? and Result
• Cancellation is supported where needed
• Resource limits are explicit and documented
• Async functions are instrumented for observability
• Tests cover timeout and cancellation scenarios

References

• User rule: rule-memory-and-ressources.md (Resource & Memory Budget)
• User rule: rule-observability.md (Observability)
• Tokio documentation
• Async Rust Book

---

Converted and distributed by TomeVault — claim your Tome and manage your conversions.