atelier/languages/rs-async.md

# Rust Async — Derived Application

> Applies Atelier's domain principles to Rust async specifically.
> Derives from `domains/` docs; introduces no new P-rules (D-063).
> See `languages/rust.md` for the language first-principles stub.

## tokio and the Async Runtime (Concurrency P5 Lock Minimization, C6 Composability)

- **`tokio` is the default async runtime:** `#[tokio::main]` for the entry; `tokio::spawn` for a task. The runtime owns the reactor, the I/O driver, and the timer.
- **`tokio::spawn` returns a `JoinHandle` like `std::thread::spawn`:** a dropped `JoinHandle` detaches (the task keeps running); `await` the handle to join. Prefer await to detach.
- **`tokio::task::JoinSet` for structured concurrency:** a set of tasks awaited together; on drop, all remaining tasks are cancelled. Mirrors `errgroup`/`TaskGroup` semantics.
- **`runtime` features are explicit:** `tokio = { version = "1", features = ["full"] }` for a binary; `["rt", "rt-multi-thread", "macros"]` for a library. Pulling `full` into a library bloats downstream.

```rust
#[tokio::main]
async fn main() {
    let mut set = tokio::task::JoinSet::new();
    for id in ["a", "b", "c"] {
        set.spawn(fetch_user(id.to_string()));
    }
    while let Some(res) = set.join_next().await {
        match res {
            Ok(Ok(u)) => println!("{}", u.name),
            Ok(Err(e)) => eprintln!("err: {e}"),
            Err(join_err) => eprintln!("panic: {join_err}"),
        }
    }
}
```

## Async Traits (Concurrency P7 Cancellation Support, C6 Composability)

- **`async fn` in traits stabilized in Rust 1.75:** `trait Repo { async fn get(&self, id: &str) -> Result<User, Error>; }`. No `async-trait` crate needed for new code on recent toolchains.
- **`Box<dyn Trait>` with async methods needs `dyn`-compatibility:** the returned future is `Pin<Box<dyn Future>>`; the compiler boxes it. For hot paths, use generics (`impl Trait`) over `dyn`.
- **`async-trait` crate for older toolchains:** macro that desugars to a `Pin<Box<dyn Future>>`. Migrate to native `async fn in trait` when the toolchain allows.
- **`Send` bounds on async traits for cross-thread spawn:** `trait Repo: Send { async fn get(&self, id: &str) -> Result<User, Error>; }` — the returned future must be `Send` to spawn on a multi-thread runtime.

```rust
trait UserRepo: Send + Sync {
    async fn get(&self, id: &str) -> Result<User, Error>;
}

struct PgRepo { pool: PgPool }
impl UserRepo for PgRepo {
    async fn get(&self, id: &str) -> Result<User, Error> {
        sqlx::query_as::<_, User>("SELECT * FROM users WHERE id = $1")
            .bind(id).fetch_one(&self.pool).await.map_err(Error::from)
    }
}
```

## Cancellation (Concurrency P7 Cancellation Support, Concurrency P8 Timeout Discipline)

- **Cancellation is cooperative via dropping the future:** `tokio::select!` drops the unselected branch, cancelling it. A dropped future stops at its next `.await` point.
- **`tokio::time::timeout` for a deadline:** `timeout(Duration::from_secs(5), op).await` returns `Ok(Ok(v))` on success, `Ok(Err(e))` on inner error, `Err(Elapsed)` on timeout. Every external `await` races against a deadline (Concurrency P8).
- **`tokio::select!` for cancel-aware waits:** `select! { res = op => res, _ = cancel => return Err(Cancelled), }`. The unselected branch is dropped, cancelling it.
- **Cancellation is not atomic:** a future dropped mid-`await` may have partial state. `Drop` runs on cancellation; clean up there (e.g., rollback a transaction).
- **Applies `concurrency/P7`:** cancellation is a first-class signal; the runtime propagates it via drop. No `CancelledError` to catch — the future is gone.

```rust
use tokio::time::timeout;
use std::time::Duration;

async fn fetch_with_timeout(url: &str) -> Result<Response, Error> {
    match timeout(Duration::from_secs(5), fetch(url)).await {
        Ok(Ok(r)) => Ok(r),
        Ok(Err(e)) => Err(e.into()),
        Err(_elapsed) => Err(Error::Timeout),
    }
}

async fn cancellable(op: impl Future<Output=()>, mut cancel: tokio::sync::oneshot::Receiver<()>) {
    tokio::select! {
        _ = op => {},
        _ = &mut cancel => println!("cancelled"),
    }
}
```

## Pin and Self-Referential Futures (Concurrency P5 Lock Minimization, C1 Correctness)

- **`async fn` returns a `Future` that is often self-referential:** the generated state machine may hold a borrow into its own stack. Such a future must be `Pin`ned to move safely.
- **`Pin<Box<T>>` to box and pin:** `Box::pin(async { ... })` returns a `Pin<Box<dyn Future>>`. The cost is a heap alloc; the win is `Send`/`dyn`-compatibility.
- **`Pin<&mut T>` for in-place polling:** `Pin::new(&mut fut)` pins a stack future; the borrow checker prevents moving it. Use for stack-allocated futures in `select!`.
- **Do not `unsafe` unpin:** `Pin::get_unchecked_mut` opts out of the pin guarantees. Application code never needs it; library code uses it for `poll` implementations.

```rust
use std::pin::Pin;

async fn boxed() -> Pin<Box<dyn std::future::Future<Output = ()> + Send>> {
    Box::pin(async {
        // self-referential state machine is safe to move once pinned
    })
}
```

## Bounded Channels and Backpressure (Concurrency P9 Bounded Queues)

- **`tokio::sync::mpsc::channel(N)` is bounded:** `send().await` blocks when full (backpressure, Concurrency P9). Unbounded `unbounded_channel()` lets the producer run ahead and OOM.
- **`tokio::sync::mpsc::Sender::try_send` for non-blocking send:** returns `Err(TrySendError::Full(v))` when full; the caller decides to drop, log, or back off. A bounded queue + `try_send` is the backpressure-aware pattern.
- **`tokio::sync::broadcast` for fan-out:** multiple receivers each get a copy; a slow receiver misses (lag). Use for telemetry, not for commands.
- **Applies `messaging/queues`:** a bounded tokio channel is an in-process broker — bounded buffer, backpressure, at-most-once handoff. The same semantics apply; the broker is local.

```rust
use tokio::sync::mpsc;

async fn producer(tx: mpsc::Sender<Job>) {
    for j in jobs() {
        if tx.send(j).await.is_err() { return; } // receiver dropped
    }
}

async fn consumer(rx: mpsc::Receiver<Job>) {
    while let Some(j) = rx.recv().await {
        process(j).await;
    }
}

let (tx, rx) = mpsc::channel::<Job>(16);  // bounded: backpressure
```

## Cross-References

- `domains/concurrency/patterns.md` — the cancellation/timeout/semaphore patterns applied here.
- `domains/concurrency/first-principles.md` — Concurrency P5 Lock Minimization, P7 Cancellation Support, P8 Timeout Discipline, P9 Bounded Queues.
- `domains/messaging/delivery-semantics.md` — at-most-once vs at-least-once framing for async retry/cancel (IDEATE-40).
- `languages/rs-ownership.md` — `Send`/`Sync` bounds on futures build on the ownership model here.
- `languages/rs-tooling.md` — `tokio` feature flags and the `cargo` build profiles detailed there.
- `languages/rs-testing.md` — `#[tokio::test]` and async test patterns.