1---
2name: rust-async-patterns
3description: 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.
4---
5 
6# Rust Async Patterns
7 
8Production patterns for async Rust programming with Tokio runtime, including tasks, channels, streams, and error handling.
9 
10## When to Use This Skill
11 
12- Building async Rust applications
13- Implementing concurrent network services
14- Using Tokio for async I/O
15- Handling async errors properly
16- Debugging async code issues
17- Optimizing async performance
18 
19## Core Concepts
20 
21### 1. Async Execution Model
22 
23```
24Future (lazy) → poll() → Ready(value) | Pending
25                ↑           ↓
26              Waker ← Runtime schedules
27```
28 
29### 2. Key Abstractions
30 
31| Concept    | Purpose                                  |
32| ---------- | ---------------------------------------- |
33| `Future`   | Lazy computation that may complete later |
34| `async fn` | Function returning impl Future           |
35| `await`    | Suspend until future completes           |
36| `Task`     | Spawned future running concurrently      |
37| `Runtime`  | Executor that polls futures              |
38 
39## Quick Start
40 
41```toml
42# Cargo.toml
43[dependencies]
44tokio = { version = "1", features = ["full"] }
45futures = "0.3"
46async-trait = "0.1"
47anyhow = "1.0"
48tracing = "0.1"
49tracing-subscriber = "0.3"
50```
51 
52```rust
53use tokio::time::{sleep, Duration};
54use anyhow::Result;
55 
56#[tokio::main]
57async fn main() -> Result<()> {
58    // Initialize tracing
59    tracing_subscriber::fmt::init();
60 
61    // Async operations
62    let result = fetch_data("https://api.example.com").await?;
63    println!("Got: {}", result);
64 
65    Ok(())
66}
67 
68async fn fetch_data(url: &str) -> Result<String> {
69    // Simulated async operation
70    sleep(Duration::from_millis(100)).await;
71    Ok(format!("Data from {}", url))
72}
73```
74 
75## Patterns
76 
77### Pattern 1: Concurrent Task Execution
78 
79```rust
80use tokio::task::JoinSet;
81use anyhow::Result;
82 
83// Spawn multiple concurrent tasks
84async fn fetch_all_concurrent(urls: Vec<String>) -> Result<Vec<String>> {
85    let mut set = JoinSet::new();
86 
87    for url in urls {
88        set.spawn(async move {
89            fetch_data(&url).await
90        });
91    }
92 
93    let mut results = Vec::new();
94    while let Some(res) = set.join_next().await {
95        match res {
96            Ok(Ok(data)) => results.push(data),
97            Ok(Err(e)) => tracing::error!("Task failed: {}", e),
98            Err(e) => tracing::error!("Join error: {}", e),
99        }
100    }
101 
102    Ok(results)
103}
104 
105// With concurrency limit
106use futures::stream::{self, StreamExt};
107 
108async fn fetch_with_limit(urls: Vec<String>, limit: usize) -> Vec<Result<String>> {
109    stream::iter(urls)
110        .map(|url| async move { fetch_data(&url).await })
111        .buffer_unordered(limit) // Max concurrent tasks
112        .collect()
113        .await
114}
115 
116// Select first to complete
117use tokio::select;
118 
119async fn race_requests(url1: &str, url2: &str) -> Result<String> {
120    select! {
121        result = fetch_data(url1) => result,
122        result = fetch_data(url2) => result,
123    }
124}
125```
126 
127### Pattern 2: Channels for Communication
128 
129```rust
130use tokio::sync::{mpsc, broadcast, oneshot, watch};
131 
132// Multi-producer, single-consumer
133async fn mpsc_example() {
134    let (tx, mut rx) = mpsc::channel::<String>(100);
135 
136    // Spawn producer
137    let tx2 = tx.clone();
138    tokio::spawn(async move {
139        tx2.send("Hello".to_string()).await.unwrap();
140    });
141 
142    // Consume
143    while let Some(msg) = rx.recv().await {
144        println!("Got: {}", msg);
145    }
146}
147 
148// Broadcast: multi-producer, multi-consumer
149async fn broadcast_example() {
150    let (tx, _) = broadcast::channel::<String>(100);
151 
152    let mut rx1 = tx.subscribe();
153    let mut rx2 = tx.subscribe();
154 
155    tx.send("Event".to_string()).unwrap();
156 
157    // Both receivers get the message
158    let _ = rx1.recv().await;
159    let _ = rx2.recv().await;
160}
161 
162// Oneshot: single value, single use
163async fn oneshot_example() -> String {
164    let (tx, rx) = oneshot::channel::<String>();
165 
166    tokio::spawn(async move {
167        tx.send("Result".to_string()).unwrap();
168    });
169 
170    rx.await.unwrap()
171}
172 
173// Watch: single producer, multi-consumer, latest value
174async fn watch_example() {
175    let (tx, mut rx) = watch::channel("initial".to_string());
176 
177    tokio::spawn(async move {
178        loop {
179            // Wait for changes
180            rx.changed().await.unwrap();
181            println!("New value: {}", *rx.borrow());
182        }
183    });
184 
185    tx.send("updated".to_string()).unwrap();
186}
187```
188 
189### Pattern 3: Async Error Handling
190 
191```rust
192use anyhow::{Context, Result, bail};
193use thiserror::Error;
194 
195#[derive(Error, Debug)]
196pub enum ServiceError {
197    #[error("Network error: {0}")]
198    Network(#[from] reqwest::Error),
199 
200    #[error("Database error: {0}")]
201    Database(#[from] sqlx::Error),
202 
203    #[error("Not found: {0}")]
204    NotFound(String),
205 
206    #[error("Timeout after {0:?}")]
207    Timeout(std::time::Duration),
208}
209 
210// Using anyhow for application errors
211async fn process_request(id: &str) -> Result<Response> {
212    let data = fetch_data(id)
213        .await
214        .context("Failed to fetch data")?;
215 
216    let parsed = parse_response(&data)
217        .context("Failed to parse response")?;
218 
219    Ok(parsed)
220}
221 
222// Using custom errors for library code
223async fn get_user(id: &str) -> Result<User, ServiceError> {
224    let result = db.query(id).await?;
225 
226    match result {
227        Some(user) => Ok(user),
228        None => Err(ServiceError::NotFound(id.to_string())),
229    }
230}
231 
232// Timeout wrapper
233use tokio::time::timeout;
234 
235async fn with_timeout<T, F>(duration: Duration, future: F) -> Result<T, ServiceError>
236where
237    F: std::future::Future<Output = Result<T, ServiceError>>,
238{
239    timeout(duration, future)
240        .await
241        .map_err(|_| ServiceError::Timeout(duration))?
242}
243```
244 
245### Pattern 4: Graceful Shutdown
246 
247```rust
248use tokio::signal;
249use tokio::sync::broadcast;
250use tokio_util::sync::CancellationToken;
251 
252async fn run_server() -> Result<()> {
253    // Method 1: CancellationToken
254    let token = CancellationToken::new();
255    let token_clone = token.clone();
256 
257    // Spawn task that respects cancellation
258    tokio::spawn(async move {
259        loop {
260            tokio::select! {
261                _ = token_clone.cancelled() => {
262                    tracing::info!("Task shutting down");
263                    break;
264                }
265                _ = do_work() => {}
266            }
267        }
268    });
269 
270    // Wait for shutdown signal
271    signal::ctrl_c().await?;
272    tracing::info!("Shutdown signal received");
273 
274    // Cancel all tasks
275    token.cancel();
276 
277    // Give tasks time to cleanup
278    tokio::time::sleep(Duration::from_secs(5)).await;
279 
280    Ok(())
281}
282 
283// Method 2: Broadcast channel for shutdown
284async fn run_with_broadcast() -> Result<()> {
285    let (shutdown_tx, _) = broadcast::channel::<()>(1);
286 
287    let mut rx = shutdown_tx.subscribe();
288    tokio::spawn(async move {
289        tokio::select! {
290            _ = rx.recv() => {
291                tracing::info!("Received shutdown");
292            }
293            _ = async { loop { do_work().await } } => {}
294        }
295    });
296 
297    signal::ctrl_c().await?;
298    let _ = shutdown_tx.send(());
299 
300    Ok(())
301}
302```
303 
304### Pattern 5: Async Traits
305 
306```rust
307use async_trait::async_trait;
308 
309#[async_trait]
310pub trait Repository {
311    async fn get(&self, id: &str) -> Result<Entity>;
312    async fn save(&self, entity: &Entity) -> Result<()>;
313    async fn delete(&self, id: &str) -> Result<()>;
314}
315 
316pub struct PostgresRepository {
317    pool: sqlx::PgPool,
318}
319 
320#[async_trait]
321impl Repository for PostgresRepository {
322    async fn get(&self, id: &str) -> Result<Entity> {
323        sqlx::query_as!(Entity, "SELECT * FROM entities WHERE id = $1", id)
324            .fetch_one(&self.pool)
325            .await
326            .map_err(Into::into)
327    }
328 
329    async fn save(&self, entity: &Entity) -> Result<()> {
330        sqlx::query!(
331            "INSERT INTO entities (id, data) VALUES ($1, $2)
332             ON CONFLICT (id) DO UPDATE SET data = $2",
333            entity.id,
334            entity.data
335        )
336        .execute(&self.pool)
337        .await?;
338        Ok(())
339    }
340 
341    async fn delete(&self, id: &str) -> Result<()> {
342        sqlx::query!("DELETE FROM entities WHERE id = $1", id)
343            .execute(&self.pool)
344            .await?;
345        Ok(())
346    }
347}
348 
349// Trait object usage
350async fn process(repo: &dyn Repository, id: &str) -> Result<()> {
351    let entity = repo.get(id).await?;
352    // Process...
353    repo.save(&entity).await
354}
355```
356 
357### Pattern 6: Streams and Async Iteration
358 
359```rust
360use futures::stream::{self, Stream, StreamExt};
361use async_stream::stream;
362 
363// Create stream from async iterator
364fn numbers_stream() -> impl Stream<Item = i32> {
365    stream! {
366        for i in 0..10 {
367            tokio::time::sleep(Duration::from_millis(100)).await;
368            yield i;
369        }
370    }
371}
372 
373// Process stream
374async fn process_stream() {
375    let stream = numbers_stream();
376 
377    // Map and filter
378    let processed: Vec<_> = stream
379        .filter(|n| futures::future::ready(*n % 2 == 0))
380        .map(|n| n * 2)
381        .collect()
382        .await;
383 
384    println!("{:?}", processed);
385}
386 
387// Chunked processing
388async fn process_in_chunks() {
389    let stream = numbers_stream();
390 
391    let mut chunks = stream.chunks(3);
392 
393    while let Some(chunk) = chunks.next().await {
394        println!("Processing chunk: {:?}", chunk);
395    }
396}
397 
398// Merge multiple streams
399async fn merge_streams() {
400    let stream1 = numbers_stream();
401    let stream2 = numbers_stream();
402 
403    let merged = stream::select(stream1, stream2);
404 
405    merged
406        .for_each(|n| async move {
407            println!("Got: {}", n);
408        })
409        .await;
410}
411```
412 
413### Pattern 7: Resource Management
414 
415```rust
416use std::sync::Arc;
417use tokio::sync::{Mutex, RwLock, Semaphore};
418 
419// Shared state with RwLock (prefer for read-heavy)
420struct Cache {
421    data: RwLock<HashMap<String, String>>,
422}
423 
424impl Cache {
425    async fn get(&self, key: &str) -> Option<String> {
426        self.data.read().await.get(key).cloned()
427    }
428 
429    async fn set(&self, key: String, value: String) {
430        self.data.write().await.insert(key, value);
431    }
432}
433 
434// Connection pool with semaphore
435struct Pool {
436    semaphore: Semaphore,
437    connections: Mutex<Vec<Connection>>,
438}
439 
440impl Pool {
441    fn new(size: usize) -> Self {
442        Self {
443            semaphore: Semaphore::new(size),
444            connections: Mutex::new((0..size).map(|_| Connection::new()).collect()),
445        }
446    }
447 
448    async fn acquire(&self) -> PooledConnection<'_> {
449        let permit = self.semaphore.acquire().await.unwrap();
450        let conn = self.connections.lock().await.pop().unwrap();
451        PooledConnection { pool: self, conn: Some(conn), _permit: permit }
452    }
453}
454 
455struct PooledConnection<'a> {
456    pool: &'a Pool,
457    conn: Option<Connection>,
458    _permit: tokio::sync::SemaphorePermit<'a>,
459}
460 
461impl Drop for PooledConnection<'_> {
462    fn drop(&mut self) {
463        if let Some(conn) = self.conn.take() {
464            let pool = self.pool;
465            tokio::spawn(async move {
466                pool.connections.lock().await.push(conn);
467            });
468        }
469    }
470}
471```
472 
473## Debugging Tips
474 
475```rust
476// Enable tokio-console for runtime debugging
477// Cargo.toml: tokio = { features = ["tracing"] }
478// Run: RUSTFLAGS="--cfg tokio_unstable" cargo run
479// Then: tokio-console
480 
481// Instrument async functions
482use tracing::instrument;
483 
484#[instrument(skip(pool))]
485async fn fetch_user(pool: &PgPool, id: &str) -> Result<User> {
486    tracing::debug!("Fetching user");
487    // ...
488}
489 
490// Track task spawning
491let span = tracing::info_span!("worker", id = %worker_id);
492tokio::spawn(async move {
493    // Enters span when polled
494}.instrument(span));
495```
496 
497## Best Practices
498 
499### Do's
500 
501- **Use `tokio::select!`** - For racing futures
502- **Prefer channels** - Over shared state when possible
503- **Use `JoinSet`** - For managing multiple tasks
504- **Instrument with tracing** - For debugging async code
505- **Handle cancellation** - Check `CancellationToken`
506 
507### Don'ts
508 
509- **Don't block** - Never use `std::thread::sleep` in async
510- **Don't hold locks across awaits** - Causes deadlocks
511- **Don't spawn unboundedly** - Use semaphores for limits
512- **Don't ignore errors** - Propagate with `?` or log
513- **Don't forget Send bounds** - For spawned futures
514