# Build a request and reply actor with fibre oneshot

**For:** Rust developers who know the basics of Tokio.

You'll build a small bank actor: one Tokio task owns the account balances and every other part of the program talks to it by sending a request with a `oneshot` reply channel inside. Along the way you'll see what a caller gets when the actor never answers, how to put a timeout on a reply, how a plain thread waits for one and how to reuse reply channels from a pool when requests come in fast.

Before you start: you need Rust with Cargo. fibre comes from crates.io as `fibre = "0"`.

## Set up the project

Start a new binary crate:

```sh
cargo new fibre-bank
cd fibre-bank
```

Add fibre and Tokio to `Cargo.toml`. The program needs Tokio's single-threaded runtime, its macros and its timers.

```toml title="Cargo.toml"
[dependencies]
fibre = "0"
tokio = { version = "1", features = ["macros", "rt", "time"] }
```

Everything runs on the `current_thread` runtime, so the output comes out in the same order every time and you can compare it line for line with what's shown here.

## Ask the actor for a value

An actor is a task that owns some state and changes it only in response to messages. Callers can't touch the state directly, so they need a way to get an answer back. The usual shape is a request queue into the actor plus a fresh one-value reply channel per request. The request queue here is an `mpsc` channel, since many callers send to one actor. Each reply is a `oneshot`.

fibre's `oneshot` module has a few constructors. `oneshot::exclusive()` is the one for plain request and reply: it gives you an `ExclusiveSender` and an `ExclusiveReceiver`, neither of which can be cloned. Sending on it is a single slot write. `oneshot::oneshot()` returns a sender you can clone, for the case where several tasks race to fill one reply and only the first `send` wins. This tutorial uses `exclusive()`.

Replace `src/main.rs` with this:

```rust title="src/main.rs"
use fibre::mpsc;
use fibre::oneshot::{self, ExclusiveSender};
use std::collections::HashMap;

enum Request {
    Deposit { account: &'static str, amount: u64, reply: ExclusiveSender<u64> },
    Balance { account: &'static str, reply: ExclusiveSender<Option<u64>> },
}

async fn run_bank(inbox: mpsc::BoundedAsyncReceiver<Request>) {
    let mut balances: HashMap<&'static str, u64> = HashMap::new();
    while let Ok(request) = inbox.recv().await {
        match request {
            Request::Deposit { account, amount, reply } => {
                let balance = balances.entry(account).or_insert(0);
                *balance += amount;
                let _ = reply.send(*balance);
            }
            Request::Balance { account, reply } => {
                let _ = reply.send(balances.get(account).copied());
            }
        }
    }
    println!("bank: all callers gone, exiting");
}

#[derive(Clone)]
struct Bank {
    requests: mpsc::BoundedAsyncSender<Request>,
}

impl Bank {
    async fn deposit(&self, account: &'static str, amount: u64) -> u64 {
        let (reply, mut answer) = oneshot::exclusive();
        self.requests.send(Request::Deposit { account, amount, reply }).await.expect("bank is running");
        answer.recv().await.expect("bank replied")
    }

    async fn balance(&self, account: &'static str) -> Option<u64> {
        let (reply, mut answer) = oneshot::exclusive();
        self.requests.send(Request::Balance { account, reply }).await.expect("bank is running");
        answer.recv().await.expect("bank replied")
    }
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let (requests, inbox) = mpsc::bounded_async::<Request>(32);
    let actor = tokio::spawn(run_bank(inbox));
    let bank = Bank { requests };

    println!("alice: deposited 50, balance {}", bank.deposit("alice", 50).await);
    println!("alice: deposited 25, balance {}", bank.deposit("alice", 25).await);
    println!("bob: balance {:?}", bank.balance("bob").await);

    drop(bank);
    actor.await.unwrap();
}
```

Each `Request` variant carries the sending half of its reply channel. The caller keeps the receiving half, sends the request and then awaits `recv()` on its receiver. The actor answers with `reply.send(value)`, which takes the sender by value, so a reply can only ever be sent once. A oneshot send never waits: the slot is always free, so there is no async version of it and no backpressure to think about.

`Bank` is the handle callers use. It wraps the request sender and hides the channel plumbing behind ordinary async methods. It's `Clone` because `mpsc` senders are. When the last `Bank` is dropped, the request queue disconnects, `inbox.recv()` returns an error and the actor loop ends. Run it:

```shell-session
$ cargo run
alice: deposited 50, balance 50
alice: deposited 25, balance 75
bob: balance None
bank: all callers gone, exiting
```

The actor ignores the result of `reply.send` with `let _ =`. A send fails only when the caller's receiver is already gone. At that point there's nobody left to tell. You'll use that result later.

## Handle an actor that never replies

A reply sender can be dropped without being used. That happens when the actor panics while handling a request or when a handler takes an early return and forgets to reply. The caller doesn't hang in any of these cases: `recv()` resolves with `Err(RecvError::Disconnected)` as soon as the sender is gone.

To see it, add a withdraw request whose handler has a gap. An unknown account hits `continue` and the whole request, reply sender included, is dropped. Add the variant to `Request`:

```rust title="src/main.rs"
    Withdraw { account: &'static str, amount: u64, reply: ExclusiveSender<Option<u64>> },
```

Add its arm to the `match` in `run_bank`:

```rust title="src/main.rs"
            Request::Withdraw { account, amount, reply } => {
                let Some(balance) = balances.get_mut(account) else {
                    println!("bank: no account {account}, skipping");
                    continue;
                };
                let left = balance.checked_sub(amount);
                if let Some(left) = left {
                    *balance = left;
                }
                let _ = reply.send(left);
            }
```

This caller method returns the `recv()` result as it is, so you can see the error. Add `use fibre::error::RecvError;` at the top and this method to `impl Bank`:

```rust title="src/main.rs"
    async fn withdraw(&self, account: &'static str, amount: u64) -> Result<Option<u64>, RecvError> {
        let (reply, mut answer) = oneshot::exclusive();
        self.requests.send(Request::Withdraw { account, amount, reply }).await.expect("bank is running");
        answer.recv().await
    }
```

Then change the calls in `main`:

```rust title="src/main.rs"
    println!("alice: deposited 50, balance {}", bank.deposit("alice", 50).await);
    println!("alice: withdraw 30 -> {:?}", bank.withdraw("alice", 30).await);
    println!("alice: withdraw 30 -> {:?}", bank.withdraw("alice", 30).await);
    println!("carol: withdraw 10 -> {:?}", bank.withdraw("carol", 10).await);
    println!("alice: balance {:?}", bank.balance("alice").await);
```

```text
alice: deposited 50, balance 50
alice: withdraw 30 -> Ok(Some(20))
alice: withdraw 30 -> Ok(None)
bank: no account carol, skipping
carol: withdraw 10 -> Err(Disconnected)
alice: balance Some(20)
bank: all callers gone, exiting
```

The second withdrawal is refused with a proper answer, `Ok(None)`. Carol's request gets no answer at all and the caller finds out from `Disconnected`. The actor keeps running and serves the next request normally. In a real program you'd fix the handler to reply for every request and treat `Disconnected` as "the actor is broken", but it's good to know the caller is never left waiting.

## Put a timeout on the reply

Sometimes the actor is alive but slow. Add a statement request that takes 200ms to build. The actor also checks the result of `send` this time: if it gets `TrySendError::Closed` back, the caller stopped waiting and the value comes back inside the error.

Replace the imports at the top with these:

```rust title="src/main.rs"
use fibre::error::{RecvError, TrySendError};
use fibre::mpsc;
use fibre::oneshot::{self, ExclusiveSender};
use std::collections::HashMap;
use std::time::Duration;
use tokio::time::{sleep, timeout};
```

Add the variant:

```rust title="src/main.rs"
    Statement { account: &'static str, reply: ExclusiveSender<String> },
```

Add its arm to `run_bank`:

```rust title="src/main.rs"
            Request::Statement { account, reply } => {
                sleep(Duration::from_millis(200)).await;
                let text = format!("statement for {account}: {:?}", balances.get(account));
                if let Err(TrySendError::Closed(_)) = reply.send(text) {
                    println!("bank: {account}'s caller gave up, statement discarded");
                }
            }
```

Add the caller method to `impl Bank`:

```rust title="src/main.rs"
    async fn statement(&self, account: &'static str) -> String {
        let (reply, mut answer) = oneshot::exclusive();
        self.requests.send(Request::Statement { account, reply }).await.expect("bank is running");
        answer.recv().await.expect("bank replied")
    }
```

In `main`, wrap the call in `tokio::time::timeout`. Replace the calls from the last step with these:

```rust title="src/main.rs"
    println!("alice: deposited 50, balance {}", bank.deposit("alice", 50).await);

    match timeout(Duration::from_millis(100), bank.statement("alice")).await {
        Ok(text) => println!("alice: {text}"),
        Err(_) => println!("alice: no statement after 100ms, giving up"),
    }
    println!("alice: balance {:?}", bank.balance("alice").await);

    match timeout(Duration::from_millis(300), bank.statement("alice")).await {
        Ok(text) => println!("alice: {text}"),
        Err(_) => println!("alice: no statement after 300ms, giving up"),
    }
    println!("carol: withdraw 10 -> {:?}", bank.withdraw("carol", 10).await);
```

```text
alice: deposited 50, balance 50
alice: no statement after 100ms, giving up
bank: alice's caller gave up, statement discarded
alice: balance Some(50)
alice: statement for alice: Some(50)
bank: no account carol, skipping
carol: withdraw 10 -> Err(Disconnected)
bank: all callers gone, exiting
```

When the 100ms timeout fires, Tokio drops the `statement` future and the `ExclusiveReceiver` inside it goes with it. Dropping a receiver closes the channel. The actor doesn't know yet: it finishes the statement at 200ms and only learns the caller left when `send` returns `Closed`. Meanwhile the balance request is queued behind the statement, so it's answered right after. The second statement has 300ms, which is enough.

A timeout on the reply doesn't cancel the work. If a request is expensive, the actor can check `reply.is_closed()` before starting and skip requests whose callers have already given up.

## Wait for a reply from a plain thread

Not every caller is async. Every oneshot receiver has `recv_blocking()` for synchronous code. The sending side needs no sync version, since a oneshot send never waits. The request queue does need a blocking sender. `to_sync()` converts an async `mpsc` sender into one.

Add this to `main`, just before `drop(bank)`:

```rust title="src/main.rs"
    let requests = bank.requests.clone().to_sync();
    let report = tokio::task::spawn_blocking(move || {
        let (reply, mut answer) = oneshot::exclusive();
        requests.send(Request::Balance { account: "alice", reply }).expect("bank is running");
        answer.recv_blocking().expect("bank replied")
    });
    println!("report thread: alice balance {:?}", report.await.unwrap());
```

```text
alice: deposited 50, balance 50
alice: no statement after 100ms, giving up
bank: alice's caller gave up, statement discarded
alice: balance Some(50)
alice: statement for alice: Some(50)
bank: no account carol, skipping
carol: withdraw 10 -> Err(Disconnected)
report thread: alice balance Some(50)
bank: all callers gone, exiting
```

`spawn_blocking` runs the closure on a separate thread, so `recv_blocking()` parks that thread and leaves the runtime free to run the actor. The actor sends the reply from async code and the thread wakes up with it. The converted sender is a clone of the request queue sender. It's dropped when the closure returns, so the actor still shuts down once `bank` is dropped.

## Reuse reply channels from a pool

Each `exclusive()` call allocates a small channel and frees it once both ends are done. At high request rates that allocation shows up. For that case fibre has `OneshotPairPool`: a fixed set of reply channels that get recycled. `pair()` hands out a `PooledSender` and `PooledReceiver` that behave like the exclusive pair. When both are finished, the slot goes back into the pool. When every slot is in use, `pair()` returns `None`, which gives you a hard cap on requests in flight.

Switch deposits to a pool. Update the oneshot import and add `Arc`:

```rust title="src/main.rs"
use fibre::oneshot::{self, ExclusiveSender, OneshotPairPool, PooledSender};
use std::collections::HashMap;
use std::sync::Arc;
```

Change the reply type of `Deposit`:

```rust title="src/main.rs"
    Deposit { account: &'static str, amount: u64, reply: PooledSender<u64> },
```

The actor's `Deposit` arm doesn't change, because a `PooledSender` has the same `send` as an `ExclusiveSender`. Give `Bank` the pool and an error for when it's empty:

```rust title="src/main.rs"
#[derive(Debug)]
struct Busy;

#[derive(Clone)]
struct Bank {
    requests: mpsc::BoundedAsyncSender<Request>,
    deposit_replies: Arc<OneshotPairPool<u64>>,
}
```

Replace `deposit` so it takes its reply channel from the pool:

```rust title="src/main.rs"
    async fn deposit(&self, account: &'static str, amount: u64) -> Result<u64, Busy> {
        let (reply, mut answer) = self.deposit_replies.pair().ok_or(Busy)?;
        self.requests.send(Request::Deposit { account, amount, reply }).await.expect("bank is running");
        Ok(answer.recv().await.expect("bank replied"))
    }
```

In `main`, build the bank with a pool of 2:

```rust title="src/main.rs"
    let bank = Bank { requests, deposit_replies: Arc::new(oneshot::pair_pool(2)) };
```

Then replace the first deposit line with three concurrent callers and a retry:

```rust title="src/main.rs"
    let mut callers = Vec::new();
    for (account, amount) in [("alice", 50), ("bob", 20), ("dave", 30)] {
        let bank = bank.clone();
        callers.push(tokio::spawn(async move {
            match bank.deposit(account, amount).await {
                Ok(balance) => println!("{account}: deposited {amount}, balance {balance}"),
                Err(Busy) => println!("{account}: no reply slot free, try again later"),
            }
        }));
    }
    for caller in callers {
        caller.await.unwrap();
    }
    println!("dave: retry -> {:?}", bank.deposit("dave", 30).await);
```

```text
dave: no reply slot free, try again later
alice: deposited 50, balance 50
bob: deposited 20, balance 20
dave: retry -> Ok(30)
alice: no statement after 100ms, giving up
bank: alice's caller gave up, statement discarded
alice: balance Some(50)
alice: statement for alice: Some(50)
bank: no account carol, skipping
carol: withdraw 10 -> Err(Disconnected)
report thread: alice balance Some(50)
bank: all callers gone, exiting
```

The three callers run in the order they were spawned. Alice and Bob each take a slot and wait for their replies. Dave finds both slots in use and gets `Busy` straight away. Once Alice and Bob have their answers, their slots are back in the pool and Dave's retry goes through.

The pool is `Send + Sync`, so one `Arc` shared by every `Bank` clone is enough. Size it for the number of requests you expect in flight at once. If a caller times out holding a pooled receiver, the slot is returned when the actor's send finishes, so a slow actor holds slots for as long as it's slow. The other request types keep using `exclusive()`. Mixing the two in one actor is fine.

## Where to go next

The channels guide at [https://github.com/excsn/fibre/blob/main/channels/README.GUIDE.md](https://github.com/excsn/fibre/blob/main/channels/README.GUIDE.md) covers the rest of the oneshot module, including `OneshotHostPool`, which pools a whole request record together with its reply channel so a request and reply allocates nothing. The same guide covers `mpsc`, `mpmc`, `spsc` and `spmc` and how each mixes sync threads with async tasks.

The full API, including every error type, is on docs.rs at [https://docs.rs/fibre](https://docs.rs/fibre).