rust-async-patterns

name	rust-async-patterns
description	Master Rust async programming with Tokio, async traits, error handling, and concurrent patterns. Use when building async Rust applications, implementing concurrent systems, or debugging async code. Use when this capability is needed.
metadata	{"author":"bl1nk-bot"}

Rust Async Patterns

Production patterns for async Rust programming with Tokio runtime, including tasks, channels, streams, and error handling.

When to Use This Skill

• Building async Rust applications
• Implementing concurrent network services
• Using Tokio for async I/O
• Handling async errors properly
• Debugging async code issues
• Optimizing async performance

Core Concepts

1. Async Execution Model

Future (lazy) → poll() → Ready(value) | Pending
                ↑           ↓
              Waker ← Runtime schedules

2. Key Abstractions

Concept	Purpose
Future	Lazy computation that may complete later
async fn	Function returning impl Future
await	Suspend until future completes
Task	Spawned future running concurrently
Runtime	Executor that polls futures

Quick Start

# Cargo.toml
[dependencies]
tokio = { version = "1", features = ["full"] }
futures = "0.3"
async-trait = "0.1"
anyhow = "1.0"
tracing = "0.1"
tracing-subscriber = "0.3"

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

#[tokio::main]
async fn main() -> Result<()> {
    // Initialize tracing
    tracing_subscriber::fmt::init();

    // Async operations
    let result = fetch_data("https://api.example.com").await?;
    println!("Got: {}", result);

    Ok(())
}

async fn fetch_data(url: &str) -> Result<String> {
    // Simulated async operation
    sleep(Duration::from_millis(100)).await;
    Ok(format!("Data from {}", url))
}
```text
## Patterns

### Pattern 1: Concurrent Task Execution
```rust
use tokio::task::JoinSet;
use anyhow::Result;

// Spawn multiple concurrent tasks
async fn fetch_all_concurrent(urls: Vec<String>) -> Result<Vec<String>> {
      = JoinSet::();

       urls {
        set.(  {
            (&url).
        });
    }

      = ::();
      (res) = set.(). {
         res {
            ((data)) => results.(data),
            ((e)) => tracing::error!(, e),
            (e) => tracing::error!(, e),
        }
    }

    (results)
}


 futures::stream::{, StreamExt};

  (urls: <>, limit: )  <<>> {
    stream::(urls)
        .(|url|   { (&url). })
        .(limit) 
        .()
        .
}


 tokio::select;

  (url1: &, url2: &)  <> {
     {
        result = (url1) => result,
        result = (url2) => result,
    }
}
```text
### Pattern : Channels  
```rust
 tokio::sync::{mpsc, broadcast, oneshot, watch};


  () {
     (tx,  rx) = mpsc::channel::<>();

    
      = tx.();
    tokio::(  {
        tx2.(.())..();
    });

    
      (msg) = rx.(). {
        (, msg);
    }
}


  () {
     (tx, _) = broadcast::channel::<>();

      = tx.();
      = tx.();

    tx.(.()).();

    
      = rx1.().;
      = rx2.().;
}


  ()   {
     (tx, rx) = oneshot::channel::<>();

    tokio::(  {
        tx.(.()).();
    });

    rx..()
}


  () {
     (tx,  rx) = watch::(.());

    tokio::(  {
         {
            
            rx.()..();
            (, *rx.());
        }
    });

    tx.(.()).();
}
```text
### Pattern : Async Error Handling
```rust
 anyhow::{Context, , bail};
 thiserror::Error;


   {
    
    ( reqwest::Error),

    
    ( sqlx::Error),

    
    (),

    
    (std::time::Duration),
}


  (id: &)  <Response> {
      = (id)
        .
        .()?;

      = (&data)
        .()?;

    (parsed)
}


  (id: &)  <User, ServiceError> {
      = db.(id).?;

     result {
        (user) => (user),
         => (ServiceError::(id.())),
    }
}


 tokio::time::timeout;

  <T, F>(duration: Duration, future: F)  <T, ServiceError>

    F: std::future::Future<Output = <T, ServiceError>>,
{
    (duration, future)
        .
        .(|_| ServiceError::(duration))?
}
```text
### Pattern : Graceful Shutdown
```rust
 tokio::signal;
 tokio::sync::broadcast;
 tokio_util::sync::CancellationToken;

  ()  <()> {
    
      = CancellationToken::();
      = token.();

    
    tokio::(  {
         {
            tokio:: {
                _ = token_clone.() => {
                    tracing::info!();
                    ;
                }
                _ = () => {}
            }
        }
    });

    
    signal::().?;
    tracing::info!();

    
    token.();

    
    tokio::time::(Duration::()).;

    (())
}


  ()  <()> {
     (shutdown_tx, _) = broadcast::channel::<()>();

      = shutdown_tx.();
    tokio::(  {
        tokio:: {
            _ = rx.() => {
                tracing::info!();
            }
            _ =  {  { (). } } => {}
        }
    });

    signal::().?;
      = shutdown_tx.(());

    (())
}
```text
### Pattern : Async Traits
```rust
 async_trait::async_trait;


   {
      (&, id: &)  <Entity>;
      (&, entity: &Entity)  <()>;
      (&, id: &)  <()>;
}

   {
    pool: sqlx::PgPool,
}


    {
      (&, id: &)  <Entity> {
        sqlx::query_as!(Entity, , id)
            .(&.pool)
            .
            .(::into)
    }

      (&, entity: &Entity)  <()> {
        sqlx::query!(
            ,
            entity.id,
            entity.data
        )
        .(&.pool)
        .?;
        (())
    }

      (&, id: &)  <()> {
        sqlx::query!(, id)
            .(&.pool)
            .?;
        (())
    }
}


  (repo: & Repository, id: &)  <()> {
      = repo.(id).?;
    
    repo.(&entity).
}
```text
### Pattern : Streams and Async Iteration
```rust
 futures::stream::{, Stream, StreamExt};
 async_stream::stream;


 ()   <Item = > {
    stream! {
           .. {
            tokio::time::(Duration::()).;
             i;
        }
    }
}


  () {
      = ();

    
     : <_> = stream
        .(|n| futures::future::(*n %  == ))
        .(|n| n * )
        .()
        .;

    (, processed);
}


  () {
      = ();

      = stream.();

      (chunk) = chunks.(). {
        (, chunk);
    }
}


  () {
      = ();
      = ();

      = stream::(stream1, stream2);

    merged
        .for_each(|n|   {
            (, n);
        })
        .;
}
```text
### Pattern : Resource Management
```rust
 std::sync::Arc;
 tokio::sync::{Mutex, RwLock, Semaphore};


  {
    data: RwLock<HashMap<, >>,
}

  {
      (&, key: &)  <> {
        .data.()..(key).()
    }

      (&, key: , value: ) {
        .data.()..(key, value);
    }
}


  {
    semaphore: Semaphore,
    connections: Mutex<<Connection>>,
}

  {
     (size: )   {
         {
            semaphore: Semaphore::(size),
            connections: Mutex::((..size).(|_| Connection::()).()),
        }
    }

      (&)  PooledConnection<> {
          = .semaphore.()..();
          = .connections.()..().();
        PooledConnection { pool: , conn: (conn), _permit: permit }
    }
}

 <> {
    pool: & Pool,
    conn: <Connection>,
    _permit: tokio::sync::SemaphorePermit<>,
}

   <> {
     (& ) {
          (conn) = .conn.() {
              = .pool;
            tokio::(  {
                pool.connections.()..(conn);
            });
        }
    }
}
```text
## Debugging Tips
```rust






 tracing::instrument;


  (pool: &PgPool, id: &)  <User> {
    tracing::debug!();
    
}


  = tracing::info_span!(, id = %worker_id);
tokio::(  {
    
}.(span));

Best Practices

Do's

• Use tokio::select! - For racing futures
• Prefer channels - Over shared state when possible
• Use JoinSet - For managing multiple tasks
• Instrument with tracing - For debugging async code
• Handle cancellation - Check CancellationToken

Don'ts

• Don't block - Never use std::thread::sleep in async
• Don't hold locks across awaits - Causes deadlocks
• Don't spawn unboundedly - Use semaphores for limits
• Don't ignore errors - Propagate with ? or log
• Don't forget Send bounds - For spawned futures

---

Source: bl1nk-bot/bl1nk-agents-manager — distributed by TomeVault.