Welcome

Welcome to Mainmatter’s Async Rust workshop!

You have written async Rust before. You know what async fn and .await do, you have used Tokio, and you have at some point stared at a program that was not doing what you told it to. This course is about the part that comes after the syntax: who owns your state, what happens when a future is dropped halfway through, and what your server does when it is asked for more than it can deliver.

Everything runs on Tokio. Futures, poll and Pin get one chapter of explanation and no exercises, because this is a workshop about using async rather than implementing it.

What you will build

You will take minidb, a small in-memory key-value store, and turn it into a networked one: concurrent, cancellable, back-pressured, shut down cleanly, and durable across a restart.

By the end it speaks a line protocol over TCP, one task owns the data and everything else asks it nicely, a slow client cannot starve a fast one, an overloaded server says so instead of falling over, and a restart picks up where the last one left off.

Each exercise is a complete, standalone copy of the library at that point in its evolution. You never have to carry a broken state forward.

Methodology

This is a hands-on workshop. Expect to spend at least half the day writing code.

Exercises are test-driven: each one ships a set of tests that describe the behaviour you are supposed to build, and your job is to make them pass. Some exercises hand you a todo!() to replace. In others there is nothing to replace: the text at the top of the file tells you what to change, and the tests show you its shape.

⚠️ Do not modify the tests. They are the specification. Change the code under test, not the test.

If you get stuck for more than ten minutes, grab a trainer. We are here to help. You can also find solutions to all exercises in the solutions branch of this repository.

Setup

You need a recent stable Rust toolchain:

rustup update stable

Clone the repository and create a branch to work on:

git clone https://github.com/mainmatter/async-rust-workshop
cd async-rust-workshop
git checkout -b my-solutions

Then install the workshop runner, the tool that walks you through the exercises:

cargo install --locked workshop-runner

The workflow

From the root of the repository, run:

wr

wr finds the first exercise you have not solved yet, compiles it, runs its tests, and either congratulates you or shows you what went wrong. It will not let you move on until the current exercise passes. Solve it, run wr again, and it opens the next one.

From chapter 3 onwards, every exercise that has a server in it also builds two binaries, so you can talk to the thing you have just written. The two that introduce a chapter rather than change the server, 03_server/00_intro and 04_state/00_intro, have no binaries and nothing to run. From inside the exercise’s directory:

cargo run                 # your server, on port 7878
cargo run --bin client    # in a second terminal

The client sends one line per request and prints the reply. nc localhost 7878 does the same job if you would rather not have a third terminal running cargo.

Building the whole workspace at once warns about several binaries sharing the names server and client. That is expected: every exercise carries its own pair.

That is the whole loop. Let’s make sure it works.

Exercise

The exercise for this section is located in 00_intro/00_welcome

What the runtime actually does

This chapter has no code to write. It exists so that the vocabulary the rest of the day leans on means the same thing to everyone in the room.

What Tokio is

The language gives you async fn, .await, and the Future trait, and then stops. There is no scheduler in std, no timer, and no way to wait on a socket. Tokio supplies those, along with async counterparts for the parts of std that would otherwise block a thread: files, TCP, channels, mutexes.

#[tokio::main] builds a runtime, runs your future on it, and blocks the thread until that future finishes. It gives you a thread per core; #[tokio::test] gives you a single thread, which is worth knowing the first time a test passes and production does not. Which pieces you get is a matter of feature flags, which is why the exercises ask for different ones as the day goes on: rt and macros throughout, time wherever something sleeps, net from chapter 3, sync from chapter 4, fs in chapter 9, and test-util for the paused clock.

A future is inert

An async fn does not run anything when you call it. It builds a value, and until something polls that value, nothing happens at all:

#![allow(unused)]
fn main() {
let future = touch(&counter);   // counter is still 0
future.await;                   // now it is 1
}

This is the first thing that surprises people arriving from JavaScript, C#, or Python, where calling an async function starts the work. In Rust the work is a value you own, and you decide when and where it runs. Everything else in this chapter follows from that.

.await is a suspension point

.await is not a call. It is a point at which this function is willing to be put down and picked up again later, possibly on another thread, possibly minutes later, possibly never.

Everything you are holding when you reach an .await is held across that gap. That is where most of the surprises in this workshop come from: a lock held across an await is a lock held for as long as the wait takes, and a !Send value held across an await makes the whole future !Send.

Concurrency is not parallelism

Two futures awaited one after the other take as long as both:

#![allow(unused)]
fn main() {
let first = slow_get(&store, &users, &alice).await;    // 50ms
let second = slow_get(&store, &users, &bob).await;     // 50ms, so 100ms in total
}

The same two handed to join! take as long as the slower one:

#![allow(unused)]
fn main() {
let (first, second) = tokio::join!(
    slow_get(&store, &users, &alice),
    slow_get(&store, &users, &bob),
);                                                     // 50ms
}

No threads were involved. Both futures are polled by the same task on the same thread, and while one is waiting the other makes progress.

The saving comes out of the waiting, which is also where it stops. join! interleaves polls, it does not add threads: two futures that compute for 50ms each and never suspend still take 100ms under it, and the second one is not polled at all until the first returns. Concurrency is about structure, parallelism is about hardware, and async Rust gives you the first whether or not you have the second. Work that does not wait needs the second, which is the next chapter.

The machinery, once

A Future is a trait with one method:

#![allow(unused)]
fn main() {
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output>
}

The runtime calls poll. The future either returns Ready(value) and is done, or returns Pending after arranging for cx.waker() to be called when there is a point in trying again. A future that returns Pending without keeping the waker will never be polled again, which is the single most common way to write a future that hangs.

Pin is there because an async fn compiles to a state machine that can hold references into itself. Pinning is the promise that the value will not move, which is what makes those references sound.

The exercise contains a Future implemented by hand, so the protocol is twenty lines you can read rather than something you take on trust. Put an eprintln! in its poll and run cargo test -- --nocapture if you would rather watch it than read it. That is the only poll in this workshop. From here on, the runtime does it, and the day is about the decisions you still have to make: who owns the state, what happens when a future is dropped halfway, and what your server does when it cannot keep up.

Where the work runs

A task is a lightweight, non-blocking unit of execution that drives one future to completion. tokio::spawn makes one, and from that moment it makes progress whether or not anybody awaits it. A future is just a value. The next chapter is about the difference.

Exercise

The exercise for this section is located in 01_runtime/00_intro

Tasks

A task is a lightweight, non-blocking unit of execution, and its job is to drive one future to completion. Lightweight because it is a small heap allocation rather than a thread; non-blocking because it occupies a worker thread only for the length of a single poll. tokio::spawn creates one from a future:

#![allow(unused)]
fn main() {
let handle = tokio::spawn(async move {
    store.get(&bucket, &key).cloned()
});

let value = handle.await.unwrap();
}

Three things change the moment you spawn.

It runs whether or not you await it. A future sitting in a variable is inert. A spawned task is in the runtime’s queue, and it makes progress as soon as there is a thread free. Awaiting the JoinHandle waits for the result; dropping the handle does not stop the task.

It has to be Send + 'static. The task may be picked up by any worker thread, and it may outlive whatever spawned it, so it cannot borrow from the caller and cannot hold anything that is not Send across an .await. Rc, RefCell, and MutexGuard from std all fall foul of this, and the compiler’s error will point at the .await rather than at the value, which takes some getting used to.

It can fail on its own. handle.await returns a Result, and the error case means the task panicked or was aborted. A panicking task does not bring the process down; it quietly stops existing, and the only way anybody finds out is by looking at its handle. Chapter 7 is about what to do with that.

Tasks are not threads

A task is a heap allocation and a state machine. Spawning one costs a few hundred bytes and no system call, which is why a Tokio server can have a hundred thousand of them and would fall over with a hundred thousand threads.

The tradeoff is that tasks are cooperative. A task only yields at an .await, so a task that does not await does not give anything else a turn. On a multi-threaded runtime that means one worker thread is stuck; if enough tasks do it, the whole runtime is. The next two exercises are the two halves of that: how to get concurrency when you want it, and what to do with work that cannot yield.

Exercise

The exercise for this section is located in 02_tasks/00_intro

Fetching things at the same time

Reading N keys in a loop takes N times as long as reading one:

#![allow(unused)]
fn main() {
for key in keys {
    values.push(slow_get(&store, &bucket, &key).await);
}
}

Every .await here is a full stop. Nothing else in this task happens until that lookup comes back, and the next one has not started.

Doing them at once

The fix is to have all the work in flight before awaiting any of it. JoinSet is the tool when the work is a set of tasks of the same shape:

#![allow(unused)]
fn main() {
let mut set = JoinSet::new();

set.spawn(async move { /* one piece of the work */ });   // Send + 'static, like any spawn
set.join_next().await;                                   // -> Option<Result<T, JoinError>>
}

JoinSet hands you results in completion order, not in the order you spawned them, which is exactly what you want when you are aggregating and exactly what you must not forget when the order matters. The exercise wants the values back in the order the keys came in, so it is worth deciding early whether to spawn with an index or to collect the handles in order instead.

Which tool for which shape

• join! for a fixed, small number of futures of different types. No spawning, no allocation, all on the current task, so no Send requirement.
• JoinSet for a dynamic number of tasks of the same type, when you want results as they arrive and the ability to abort them all at once.
• FuturesOrdered and FuturesUnordered, from the futures crate rather than Tokio, for the same thing without spawning, so the futures stay on this task. Cheaper, but no parallelism and no isolation from a panic.
• A Vec<JoinHandle<T>> when you want results strictly in the order you started the work.

What it costs

Spawning means the work can move to another thread, so everything it touches has to be Send + 'static. That is why the store arrives as an Arc<Store> here rather than a &Store: the task may outlive the function that spawned it, and the compiler will not let you promise otherwise.

The other cost is that N concurrent lookups are N concurrent lookups. Turning a sequential loop into an unbounded fan-out is how a service that was polite to its database becomes the reason the database is down. Chapter 6 puts a limit on it.

Exercise

The exercise for this section is located in 02_tasks/01_spawn

Work that will not yield

Async Rust is cooperative. A task keeps its worker thread until it hits an .await that returns Pending, and a task that never does that never gives the thread back:

#![allow(unused)]
fn main() {
pub fn checksum(store: &Store) -> u64 {
    // a few hundred milliseconds of pure computation, no awaits anywhere
}
}

Call that from an async function and the worker thread running it stops polling anything else for as long as it takes. On the current-thread runtime, that is the entire server. On the multi-threaded runtime it is one worker out of however many cores you have, which is worse in a way, because it shows up as a service that is fine in testing and stalls under load.

The symptom is distinctive: everything gets slower at once, including work that has nothing to do with the slow part, and a heartbeat task that should tick every ten milliseconds stops ticking.

spawn_blocking

Tokio keeps a second, much larger pool of threads for exactly this:

#![allow(unused)]
fn main() {
tokio::task::spawn_blocking(move || /* the work that will not yield */)
    .await   // -> Result<T, JoinError>
}

The closure runs on a blocking thread, the calling task awaits the result and yields while it waits, and the runtime’s workers keep serving everybody else. The pool is large (512 threads by default) because those threads are expected to be blocked most of the time.

spawn_blocking needs an owned, 'static closure, so the store arrives as an Arc<Store> again.

What counts as blocking

Anything that can take more than a moment and does not .await:

• CPU-bound work: hashing, compression, serialising something large, image processing.
• Synchronous file I/O. This is why tokio::fs exists, and why it is a wrapper around spawn_blocking rather than anything cleverer. Chapter 9 comes back to that.
• Any library that talks to the network or the disk without being async, which is most C bindings.
• std::thread::sleep, which is never what you want inside an async function. tokio::time::sleep is.

The rules of thumb

A rough line: if a piece of work can take longer than about a hundred microseconds and cannot yield, it does not belong on a runtime thread.

For genuinely CPU-heavy work that is the point of your service, spawn_blocking is a blunt instrument: its pool is sized for threads that sit waiting on I/O, not for saturating cores. The usual answer is a pool sized to the cores instead, and rayon, a data parallelism crate with its own work-stealing scheduler, is the common choice: you hand it the job and it answers on a oneshot channel back into async code. The principle is the same either way: the runtime’s threads exist to poll futures, and anything else you ask them to do is time they are not doing that.

Exercise

The exercise for this section is located in 02_tasks/02_blocking

A server

Everything so far has been minidb in one process talking to itself. From here it is a server, and it stays one for the rest of the day.

The protocol

One request per line, one response per line, all of it text:

SET users alice hello        ->  OK
GET users alice              ->  VALUE hello
GET users bob                ->  NIL
DEL users alice              ->  OK
PING                         ->  ERR unknown verb PING

A text protocol is a workshop’s best friend. You can drive the whole server with nc localhost 7878 and read every byte that goes past, which means a failure is something you can look at rather than something you have to instrument. Real systems pick differently: length-prefixed binary framing avoids the delimiter problem entirely, and gRPC or Postgres wire format arrive with tooling. The concurrency lessons are identical either way.

The one thing a line protocol needs from its types is a guarantee that a value cannot contain the delimiter, which is why Value::parse rejects newlines and anything over 4 KiB. Parse at the edge and the rest of the program cannot produce a line that does not round trip. src/protocol.rs has the test that proves it.

Reading lines

Three pieces do the framing, and it is worth knowing each of them by name rather than as a block to copy:

#![allow(unused)]
fn main() {
tokio::io::split(stream);        // -> (ReadHalf<S>, WriteHalf<S>)
BufReader::new(reader).lines();  // -> Lines<BufReader<R>>
lines.next_line().await;         // -> io::Result<Option<String>>, no trailing newline
writer.write_all(bytes).await;   // and the newline is yours to add back
}

BufReader matters for more than speed here. Without it, every read is a syscall, and with a lines() wrapper on top it is a syscall per byte. It matters again in chapter 5 for a reason that has nothing to do with performance: the buffer is what makes next_line safe to cancel.

tokio::io::split gives you a reader half and a writer half of the same stream, so a future reading and a future writing can exist at the same time. For a TcpStream specifically, into_split gives you owned halves that can be moved into separate tasks.

The binaries

From this chapter on, every exercise that changes the server carries two of them. This one does not: it is the protocol and nothing else, so there is nothing here to run. From the next exercise onwards:

cargo run                 # the server, on 127.0.0.1:7878
cargo run --bin client    # a second terminal

The client sends one line per request and prints what comes back. It is worth actually running: the tests tell you the behaviour is right, and typing at the thing tells you what it feels like.

Exercise

The exercise for this section is located in 03_server/00_intro

The accept loop

minidb is not a server yet. serve and handle_connection are both todo!(), and this is the exercise that makes them real.

A TCP server is a loop around one call:

#![allow(unused)]
fn main() {
listener.accept().await;   // -> io::Result<(TcpStream, SocketAddr)>
}

accept yields until a client turns up, hands you a TcpStream and the address it came from, and is ready to be called again. Around it goes a loop; inside it goes handle_connection, which reads lines with the three calls from the previous page, parses each one into a Request, applies it to the store, and writes the Response back.

One at a time

Awaiting handle_connection inside the loop serves exactly one client. The second one connects, the kernel holds it in the backlog queue, and nothing else happens until the first one hangs up.

That is not a bug in this exercise, it is the exercise. The single-client version is short enough to hold in your head, and it makes the next step, one task per connection, a change of two lines rather than a rewrite. It also makes the ownership obvious: with one client at a time, &mut Store is enough, and the moment there are two clients it is not.

Errors that are not errors

A client that goes away is not a failure of your server. next_line returning Ok(None) is a clean end of stream, and a write failing with BrokenPipe means somebody closed a laptop. Ending the connection quietly is the right response to both.

What you must not do is let one connection’s error end the accept loop. In this exercise handle_connection returns its error to serve, which returns it to main, which means a client that disconnects rudely takes the server down with it. The next exercise fixes that as a side effect of spawning, and chapter 7 is about telling the difference between an error that should end a connection and one that should end the process.

Ports in tests

The tests bind 127.0.0.1:0 and ask the listener which port it got:

#![allow(unused)]
fn main() {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
}

Port zero means “whatever is free”. A test suite with a hardcoded port fails when two of its own tests run at once, and fails again on the machine where something else already has 7878. This is the cheapest habit in this chapter and the one most often skipped.

Exercise

The exercise for this section is located in 03_server/01_accept

One task per connection

The loop you have serves one client at a time, because it waits for the connection it just accepted before it accepts another:

#![allow(unused)]
fn main() {
let (stream, _) = listener.accept().await?;
handle_connection(stream, &mut store).await?;   // nothing else happens until this returns
}

Serving two clients at once is a two-line change, and the tool is the one from chapter 2:

#![allow(unused)]
fn main() {
tokio::spawn(future);   // -> JoinHandle<T>, and `future` must be Send + 'static
}

The accept loop then goes straight back to accepting, and each connection makes progress on a task of its own. This is the shape of essentially every Tokio server, from this one to hyper.

Notice what spawning bought for free. An error on one connection now ends that task and nothing else, because the Result is swallowed at the boundary rather than propagated into serve. That is the right default: a connection’s problems belong to that connection.

And what it took away

handle_connection cannot borrow the store any more. tokio::spawn requires Send + 'static, and a &mut Store is neither. There is no way to spawn a task that borrows a local, no matter how obviously the local outlives it, because the compiler cannot see that and the runtime does not promise it.

So for now each connection gets a Store of its own. That compiles, and it is worse than it looks: SET users alice hello is answered OK, and the next client to ask for alice is told NIL, because the write went into a HashMap that dies with the connection that made it. A server that loses your data and says OK is worth meeting once, deliberately, and it is precisely the question chapter 4 answers: who owns the state when every connection is its own task?

Proving it, rather than hoping

The test opens a second connection while the first one is still open, and puts a deadline on the answer:

#![allow(unused)]
fn main() {
let answered = timeout(Duration::from_secs(5), second.request("SET users bob hi")).await;
}

The one-client-at-a-time server from the last exercise never answers that, because it is still inside handle_connection for the first client and stays there until that client hangs up. This is how to test for concurrency rather than hope for it, and the deadline is the part worth copying: the test fails in five seconds with a message rather than hanging forever, which is a courtesy worth extending to your own suites.

Exercise

The exercise for this section is located in 03_server/02_concurrent

Who owns the state

Every connection is a task, and every task wants the same Store. Rust will not let more than one of them have &mut Store, and it is right not to. There are two answers, and this chapter builds both.

Share it: Arc<Mutex<Store>>

#![allow(unused)]
fn main() {
let store = Arc::new(Mutex::new(Store::new()));

let mut guard = store.lock().await;
guard.insert(bucket, key, value);
}

Arc gives every task a handle to the same allocation; the mutex makes sure only one of them is inside at a time. It is the direct translation of what you would write with threads, and for a lot of services it is the correct answer.

The trap is which mutex. std::sync::Mutex is fine to use in async code as long as its guard never crosses an .await, and using it that way is often the fastest option, since the lock is uncontended and never held for long. Hold that guard across an await and you have a problem the compiler will describe badly: MutexGuard is !Send, so the whole future becomes !Send, so tokio::spawn refuses it, and the error points at the spawn rather than at the lock.

tokio::sync::Mutex is the one whose guard may be held across an await. It is slower, because waiting on it means parking a task rather than spinning, and reaching for it is often a sign that the critical section is bigger than it should be.

The other trap has nothing to do with types. A lock held while doing I/O serialises every task in the process on that I/O, and no amount of async makes that faster.

Give it away: the actor

The alternative is to stop sharing. One task owns the Store outright, and everybody else sends it messages:

#![allow(unused)]
fn main() {
pub struct StoreHandle {
    commands: mpsc::Sender<Command>,
}

struct Command {
    request: Request,
    reply: oneshot::Sender<Response>,
}
}

There is no lock, because there is nothing to lock: exactly one task ever touches the data, so it can hold &mut Store for as long as it likes. Callers get a cheap Clone handle and await their answer.

What this buys is more than tidiness:

• The queue is a place to put policy. A bounded mailbox is backpressure (chapter 6), the depth is a metric worth watching, and a request can be refused before it is queued.
• It is testable. The store task has one input and one output, and both are channels.
• It composes with cancellation. Dropping the caller drops the oneshot, and the store task can see that nobody is listening any more.

What it costs is a round trip per request, an allocation per reply, and a single task that is now a bottleneck and a single point of failure. Chapter 7 is about the second of those.

Which one

Use the mutex when the critical section is small, synchronous, and contention is low. Use the actor when the state has behaviour of its own, when you want a queue you can reason about, or when the work under the lock would otherwise involve an .await.

The exercises build both, and the chapter ends with a criterion benchmark that measures them under contention, because the honest answer to which is faster is “measure it, and the answer will change with the shape of your load”.

Exercise

The exercise for this section is located in 04_state/00_intro

A store every connection can reach

The server from chapter 3 gives every connection a store of its own, so no client can see what any other one wrote. Give them one store between them:

#![allow(unused)]
fn main() {
pub async fn serve(listener: TcpListener, store: Arc<Mutex<Store>>) -> io::Result<()> {
    loop {
        let (stream, _) = listener.accept().await?;
        let store = Arc::clone(&store);

        tokio::spawn(async move {
            let _ = handle_connection(stream, &store).await;
        });
    }
}
}

Arc::clone before the async move block is the idiom, and it reads oddly until you have written it twenty times. The clone is what the task takes ownership of; the original stays behind for the next turn of the loop.

Inside the connection

serve is written for you. handle_connection is not, and it now has a shared store where it used to have an exclusive one. One call gets you in:

#![allow(unused)]
fn main() {
store.lock().await;   // -> MutexGuard<'_, Store>, once whoever holds it lets go
}

Where that call goes is the decision. Take the lock once per request rather than once per connection, and be finished with the guard before the write, so the lock is held for the length of a HashMap operation rather than for the length of a network write to a client that may be on the other side of the world.

That is the whole discipline with a shared mutex, and it is worth stating as a rule: take the lock, do the work on the data, drop the guard, then do the I/O. A server that writes to a socket while holding the store lock has one client’s connection speed setting the throughput of every other client.

The dereference

apply wants a &mut Store, and lock().await gives you a MutexGuard<Store>, so &mut on it is a &mut MutexGuard and the types do not line up. &mut * is the operator that derefs through the guard. The compiler’s message is clear enough once you have seen it once, which is the point of meeting it here.

Why tokio::sync::Mutex here

Follow the rule above and the guard never reaches an .await, so std::sync::Mutex would work here and would be faster: it is a plain lock with no task parking, and an uncontended one costs an atomic swap. The exercise uses Tokio’s anyway, and it is worth being honest about why. It is the one that still compiles if you get the discipline wrong, so it is the one that lets this chapter be about where the lock goes rather than about which lock it is.

Reach for tokio::sync::Mutex when the critical section genuinely has to await something, and for std::sync::Mutex otherwise. What is not defensible is std::sync::Mutex with the guard alive across an await: the guard is not Send, so the future is not either, and tokio::spawn refuses it with an error that points at the spawn rather than at the lock. The previous exercise ships a compile_fail doctest of exactly that, in 04_state/00_intro.

Exercise

The exercise for this section is located in 04_state/01_mutex

An actor

Stop sharing the store. Give it to a task, and hand everybody else a way to ask. The two types the exercise ships say the whole design:

#![allow(unused)]
fn main() {
#[derive(Clone)]
pub struct StoreHandle {
    commands: mpsc::Sender<Command>,
}

pub struct Command {
    pub request: Request,
    pub reply: oneshot::Sender<Response>,
}
}

The task behind that sender takes the store by value and holds &mut to it for as long as it likes, because nothing else in the process can reach it. No lock, no Arc, no guard to think about.

The two channels

mpsc carries requests in. Many senders, one receiver, and the receiver is the store task. Its capacity is a decision with consequences, which is chapter 6.

oneshot carries the answer back. One value, one direction, allocated per request. The caller makes the pair, sends the sending half away inside the command, and awaits the other half.

Four calls, and what each of them reports when things go wrong:

#![allow(unused)]
fn main() {
mpsc::channel(capacity);   // -> (Sender<T>, Receiver<T>); the Sender is Clone
inbox.recv().await;        // -> Option<T>; None once the last Sender has been dropped
oneshot::channel();        // -> (Sender<T>, Receiver<T>) for exactly one value
reply.send(value);         // -> Result<(), T>; Err when nobody is waiting any more
}

Two of those can tell you the store task is gone: sending a command can fail because the channel is closed, and awaiting the answer can fail because the task died holding the oneshot sender. Both mean the same thing, neither can be ignored, and the answer to both is to tell the client the truth. Chapter 7 asks whether the server should stay up at all in that state.

What changed for the caller

Nothing, which is the point. handle_connection calls store.apply(request).await and gets a Response, exactly as it called apply under a lock. The handle is Clone, cheaply, because cloning an mpsc::Sender is what “give this task access” means.

Measuring it

The exercise ships a criterion benchmark comparing the mutex and the actor under contention:

cargo bench

It is not graded, and the number is not the lesson. What is worth doing is changing the shape of the load and watching which way the answer moves. TASKS and REQUESTS at the top of benches/store.rs are the two knobs: raise the first for more contention, the second for a longer run per task. The actor pays a message round trip per request and wins when the alternative is tasks queueing on a lock; the mutex wins when the critical section is tiny and contention is low.

Expect the mutex to win as it ships, because this workload is the one that suits it: the critical section is a single HashMap insert, which is about as small as a critical section gets. That is a real result rather than a rigged one, and it is why “use the actor” is not the moral of this chapter.

Benchmarks in async code are easy to get wrong. criterion measures wall-clock time around a block you give it, so the block has to include the runtime work you care about and nothing else. Note what the benchmark therefore does not do: the store, and the task that owns it, are built once outside the timed block, because a benchmark that allocates a channel and spawns a task on every iteration is partly measuring how fast Tokio can start things. The fan-out stays inside, because that is the workload. Treat the result as a direction, not a fact.

Exercise

The exercise for this section is located in 04_state/02_actor

Cancellation

In most languages, cancelling work means asking it to stop and hoping it checks. In Rust it means dropping a future, and it happens immediately and everywhere:

#![allow(unused)]
fn main() {
tokio::select! {
    response = store.apply(request) => response,
    _ = sleep(limit) => Response::Error("busy".to_owned()),
}
}

When the sleep wins, the other future is dropped where it stands. Not signalled. Not asked. Dropped, mid-await, with whatever it was holding.

This is the best and the sharpest thing about async Rust. Best, because cancellation is free and composable: timeout, select!, and JoinHandle::abort all work on any future, without that future having been written to support them. abort is the one that cancels a spawned task; dropping its JoinHandle detaches the task, as chapter 2 said, and detaching is not cancelling. Sharpest, because a future that is dropped at an awkward moment leaves the world in whatever state it had reached.

Where the state goes

Dropping a future runs Drop for everything the state machine holds, so memory and locks and file handles are all released correctly. What is not automatic is anything that had already reached the outside world:

• Bytes already written to a socket have been written. Half a response is a real thing a client can receive.
• A row already inserted stays inserted. There is no rollback unless you wrote one.
• Bytes already read out of a socket into a buffer inside the dropped future are gone, and this is the one that catches people. It is the subject of the second exercise.

The three tools

timeout(duration, future) wraps a future and drops it if it takes too long. The result is a Result<T, Elapsed>, and it is worth being precise about what Err(Elapsed) means: it means you stopped waiting, not that the work stopped happening. If the work was a message to another task, that task is still going to do it.

select! polls several futures and takes the first one that finishes, dropping the rest. Every branch is a cancellation point for the others.

CancellationToken from tokio-util is cooperative cancellation for cases where dropping is not enough, typically because you want many tasks to stop at a point of their own choosing. Chapter 7 uses it.

Cancel safety

A future is cancel safe if dropping it part-way loses nothing that cannot be recovered by calling it again. This is a property of the API you are calling, not of your code, and Tokio documents it per method under a “Cancel safety” heading.

AsyncBufReadExt::next_line is cancel safe: the bytes it has read live in the BufReader, which is yours and outlives the call. A read loop that accumulates into a local Vec is not: the Vec is inside the future, and dropping the future drops the bytes.

The rule to take home: before putting a call in a select! branch, check whether its documentation says it is cancel safe. If it does not say, or if you wrote it, assume it is not.

Exercise

The exercise for this section is located in 05_cancellation/00_intro

An idle timeout

A connection that says nothing is not free. It holds a task, a buffer, a file descriptor, and a slot in whatever limit you set later. The loop you have waits for the client for as long as the client feels like taking:

#![allow(unused)]
fn main() {
requests.next_line().await?   // -> Some(line), eventually, or never
}

Servers close those connections, and the tool for it is a macro that waits for the first of several futures rather than for one:

#![allow(unused)]
fn main() {
tokio::select! {
    a = first  => /* first finished, `second` was dropped where it stood */,
    b = second => /* and the other way round */,
}
}

select! polls every branch and takes whichever finishes first. The branch that did not win has its future dropped, which is the entire subject of this chapter: a read that loses to a timer is a read that never happened.

The two ways a connection ends

They are different and both have to be handled:

• next_line() returns Ok(None): the client hung up politely.
• The sleep fires: the client is still connected and has said nothing for idle.

Returning Ok(()) for both is right. Neither is an error, and neither deserves a log line above DEBUG.

select! is a macro with sharp edges

Three things about it are worth knowing before you need them:

Every branch is polled every time. That means every branch’s expression is evaluated on every iteration, including sleep(idle), which builds a new timer each time round the loop. Sometimes that is what you want; for a deadline it is a bug, which the next exercise makes you fix.

A branch that is not taken has its future dropped. See the previous page, and the next one.

Branch order does not decide the winner. select! polls in a random order by default, specifically so that a branch that is always ready cannot starve the others. If you need priority, biased; turns the randomisation off, and then a hot branch really can starve the rest.

Testing it without waiting

The limit is thirty seconds and the test suite runs in milliseconds:

#![allow(unused)]
fn main() {
#[tokio::test(start_paused = true)]
async fn a_client_that_says_nothing_is_hung_up_on() {
}

With the clock paused, the runtime advances time itself as soon as every task is waiting on a timer. Thirty seconds pass instantly, and Instant::now() inside the test agrees that they did. Chapter 8 is about this and what else it makes possible.

One habit from these tests: wrap the thing you expect to finish in a timeout anyway.

#![allow(unused)]
fn main() {
timeout(IDLE * 2, client.lines.next_line()).await.expect("the server never hung up")
}

A test that fails by hanging tells you nothing and blocks CI. A test that fails with a message tells you what it wanted.

Exercise

The exercise for this section is located in 05_cancellation/01_select

Cancel safety

handle_connection has grown a housekeeping branch that wakes up every TICK and goes straight back to waiting, and it reads its lines by hand, a byte at a time, into a buffer of its own:

#![allow(unused)]
fn main() {
async fn read_line_by_hand<R>(reader: &mut R) -> io::Result<Option<String>> {
    let mut line = Vec::new();
    loop {
        let mut byte = [0u8; 1];
        reader.read_exact(&mut byte).await?;
        if byte[0] == b'\n' {
            return Ok(Some(String::from_utf8_lossy(&line).into_owned()));
        }
        line.push(byte[0]);
    }
}
}

Both of those are reasonable-looking code. Together they lose data.

A client sends GET users al, pauses, and the tick fires. select! drops the read future, and the Vec holding those twelve bytes goes with it. The client sends ice\n, the next read starts from scratch, and the server answers with a complaint about a verb called ICE.

Nothing was written down anywhere that survives the drop, which is the whole definition of a future that is not cancel safe.

Why next_line is different

#![allow(unused)]
fn main() {
let mut requests = BufReader::new(reader).lines();
let line = requests.next_line().await?;
}

The bytes read so far live in the BufReader, which you own, and which is alive for the whole connection. Dropping the next_line future loses nothing, because the partial line is not in the future. Call it again and it picks up where it stopped.

That is the general shape of the fix. Move the state out of the future and into something that outlives it. Every cancel-safe API in Tokio does this, and it is how to make your own: take &mut self on a type that holds the partial state, rather than accumulating into a local.

The deadline has the same disease

Look at the third branch:

#![allow(unused)]
fn main() {
_ = sleep(idle) => return Ok(()),
}

select! evaluates every branch expression on every iteration, so this builds a new thirty-second sleep each time round the loop. With a tick every five seconds, the deadline restarts before it can ever fire, and the idle timeout from the previous exercise silently stops working. No error, no warning, and the only reason you know is that a test says so.

A future that has to outlive the iteration has to be built outside it, and then polled in place rather than consumed. Two things make that possible:

#![allow(unused)]
fn main() {
tokio::pin!(deadline);                          // Sleep -> Pin<&mut Sleep>, and it cannot move again
deadline.as_mut().reset(Instant::now() + d);    // pushes it out, without allocating a new timer
}

tokio::pin! puts the sleep somewhere it cannot move, which is what lets a branch poll it by &mut instead of taking ownership of it. Sleep::reset is what you reach for when the deadline should start again, and the question the exercise asks is when that is.

This is the one place in the workshop where Pin shows up in code you write, and the reason is exactly the one from chapter 1: a future may hold references into itself, so polling it repeatedly requires promising it will not move.

The checklist

Before a call goes in a select! branch:

1. Does its documentation have a “Cancel safety” section, and what does it say?
2. If it is yours, where does the partial state live? In the future, or in something that outlives it?
3. Is the future being created in the branch, or polled there? A deadline created in the branch is not a deadline.

Exercise

The exercise for this section is located in 05_cancellation/02_cancel_safe

Bounding the work

The idle timeout protects against a client that says nothing. It does nothing about a store that takes too long to answer, and the line that asks the store has no upper bound on it at all:

#![allow(unused)]
fn main() {
store.apply(request).await   // back when the store is ready, and not before
}

The tool wraps a future and gives up on it:

#![allow(unused)]
fn main() {
timeout(duration, future).await;   // -> Result<T, Elapsed>
}

Now every request has an upper bound, and a client gets an answer either way. That is worth something on its own: a client waiting forever cannot retry, cannot fail over, and usually cannot tell the difference between slow and dead.

What the timeout did not do

It did not stop the work.

store.apply(request) sent a message to the store task and waited for the reply. The timeout drops the waiting, so the oneshot receiver goes away, but the Command is still in the mailbox and the store task is still going to apply it. The client is told ERR busy, and the write happens anyway, a moment later, with nobody listening.

The exercise ships a test that asserts exactly this, because it is the sort of thing that is obvious once stated and invisible otherwise. It asks the store directly, since asking down the connection would time out as well:

#![allow(unused)]
fn main() {
assert_eq!(client.request("SET users alice hello").await, "ERR busy");
assert_eq!(store.apply(get()).await, Response::Value(..));
}

This is not a flaw in timeout. It is what cancellation means when the work is happening somewhere else: you can stop waiting for a message, but you cannot un-send it.

What to do about it

Three honest options, and the right one depends on what the work is:

Accept it. For an idempotent write like SET, applying it late is harmless. This is what minidb does, and for a key-value store it is defensible.

Do not queue it in the first place. If the mailbox is full, refuse before sending. That is load shedding, and it is the next chapter.

Make the work itself cancellable. Pass something the worker can check, a CancellationToken alongside the request, and have the store task drop work whose caller has gone. oneshot::Sender even has is_closed and closed, so the store task can ask whether anybody is still waiting before starting anything expensive.

Timeouts are a system property

One last thing worth saying out loud, because it is where timeouts usually go wrong in production: your timeout should be shorter than your caller’s. If a client gives up after two seconds and your server gives up after ten, you spend eight seconds doing work for somebody who has already left, and under load that is most of what you do. Timeouts that are not ordered across a call chain amplify an overload instead of shedding it.

Exercise

The exercise for this section is located in 05_cancellation/03_timeout

Backpressure

A queue with no limit is not a queue, it is a memory leak with good manners.

minidb’s store task can apply so many requests per second. If clients ask for more than that, the extra has to go somewhere, and an unbounded mpsc channel says “in RAM, all of it”. The service looks fine for a while: latency climbs, the queue grows, memory grows, and then either the OOM killer arrives or the queue drains hours of work nobody is waiting for any more.

Backpressure is the opposite arrangement: when a stage cannot keep up, the pressure travels back up the pipeline to whoever is producing, and the producer slows down or is told no.

Where it comes from in Tokio

Bounded channels give it to you almost for free:

#![allow(unused)]
fn main() {
let (commands, inbox) = mpsc::channel(MAILBOX);
}

Sender::send on a full channel waits until there is room. The task calling it makes no progress until the store task has caught up, which is the pressure, travelling backwards. That connection stops reading from its socket, its TCP receive window fills, and the client’s own write starts to block. The chain reaches all the way to the other machine without anybody writing a line of code to make it happen.

Every real system has this property somewhere, and the question is only whether you chose where.

Waiting, refusing, or dropping

Three responses to “full”, and they are not interchangeable:

• Wait (send().await). Correct when the producer has nowhere better to be and the work must happen. Latency grows, nothing is lost.
• Refuse (try_send, then tell the client). Correct when the request has a deadline anyway. This is load shedding: latency stays bounded for the requests you do accept, and the client finds out immediately.
• Drop (oldest, newest, or by priority). Correct for data where fresh matters more than complete: metrics, sensor readings, progress updates.

The first exercise is refusing. The second is a different lever entirely: limiting how many connections exist at all, so the queue is not the only thing standing between a burst and the heap.

The number

MAILBOX = 32 is a guess, and so is every other capacity in every system you have worked on. What makes it a defensible guess is knowing what it means: the queue length is latency, at the rate the consumer drains it. Thirty-two requests at a millisecond each is thirty-two milliseconds of queueing delay when full, which is a sentence you can check against your latency budget.

The queue depth is also the single most useful thing to put on a dashboard. A queue that is occasionally deep is absorbing bursts, which is its job. A queue that is permanently deep is a consumer that is too slow, and no capacity will fix it.

Exercise

The exercise for this section is located in 06_backpressure/00_intro

Refusing rather than queueing

Waiting for a full mailbox is the right default, and it is the wrong answer for a request that has a deadline anyway. REQUEST_LIMIT is two seconds; if the queue is already full of work that will take longer than that, the client is better off being told now. What apply does today is wait:

#![allow(unused)]
fn main() {
self.commands.send(command).await   // -> back when there is room, however long that takes
}

try_apply is apply with one line changed. It still makes a oneshot pair of its own and still awaits the reply on it; what differs is how the command is handed over:

#![allow(unused)]
fn main() {
self.commands.try_send(command);   // -> Result<(), TrySendError<T>>, right now, either way
}

try_send returns instead of waiting, and its two errors mean opposite things. Full is temporary and the client should try again. Closed means the store task is gone and trying again will not help anybody. Collapsing them into one answer is the kind of shortcut that turns a five-minute incident into an hour of confusion.

Shedding is a real decision

ERR busy is the same string the timeout produces, and deliberately so: from the client’s side both mean “the store is not keeping up”. What is different is the cost. A shed request costs nothing, and a timed-out one costs two seconds of a connection task’s time and, as the last chapter showed, gets applied anyway.

The trade is worth stating plainly. Shedding early keeps latency bounded for the requests you do accept, and it converts a queue that would have absorbed a burst into errors the client can see. A store that only ever sheds is a store that is too small, and a store that never sheds under any load has a queue that is too long.

Whatever you choose, count it. A shed request that is not in a metric is a service that looks healthy while refusing half its traffic.

Testing something that only happens under load

The test spawns eight requests at a store with room for one and half a second per request, and asserts that some were refused and some were accepted:

#![allow(unused)]
fn main() {
let store = StoreHandle::spawn_with_capacity(Store::new(), SLOW, 1);
}

The assertion is deliberately not “exactly three were shed”. With a paused clock the scheduling is deterministic today, and pinning the exact number would make the test a hostage to Tokio’s internal ordering. Asserting the property, that shedding happens and that it does not shed everything, is what you actually mean.

Exercise

The exercise for this section is located in 06_backpressure/01_bounded

Admission control

The mailbox is bounded and the number of connections is not. Every accepted socket is a task, a buffer, and a file descriptor, and serve will happily accept ten thousand of them. Nothing in the loop counts:

#![allow(unused)]
fn main() {
let (stream, _) = listener.accept().await?;   // ... and again, and again
}

Semaphore holds a fixed number of permits:

#![allow(unused)]
fn main() {
Semaphore::new(limit);                            // an Arc<Semaphore> is what tasks share
Arc::clone(&permits).acquire_owned().await;       // -> Result<OwnedSemaphorePermit, AcquireError>
}

Take one before serving a connection, hold it until the connection ends, and the count of live connections cannot exceed the limit. acquire_owned rather than acquire because the permit has to be moved into the task, and the permit is released by its own Drop rather than by any call.

Where you acquire it is the design

Acquire before accept and the listener stops taking connections off the kernel’s backlog when it is at capacity. A client that cannot be served yet waits in the backlog queue, which is the operating system’s memory rather than yours, and if the backlog fills the kernel refuses the connection outright. That is a fast, cheap “no” that never reaches your process.

Acquire after accept and you have accepted a connection you cannot serve. Now it is your socket, your task, and your memory, waiting for a permit, and you have moved the queue from the kernel into your heap. That is the version that looks fine and falls over.

The general shape: refuse work as early as you can, at the outermost edge where you still know enough to refuse it.

Why acquire_owned

Semaphore::acquire returns a guard that borrows the semaphore, which cannot be moved into a spawned task. acquire_owned takes an Arc<Semaphore> and returns a permit that owns its share, so it can go into the async move block.

The permit’s Drop is the whole release mechanism. There is no release call, so the slot comes back exactly when the permit dies, which means a connection that ends any way at all, including by panicking, returns its permit.

That makes where the permit lives the only thing that matters, so hand it to handle_connection:

#![allow(unused)]
fn main() {
pub async fn handle_connection<S>(
    stream: S,
    store: &StoreHandle,
    idle: Duration,
    _permit: OwnedSemaphorePermit,   // held for the length of the call, dropped when it returns
) -> io::Result<()>
}

Nothing in the body uses it, hence the underscore prefix, which silences the warning while leaving it an ordinary binding. Written this way the signature states the rule the code was only implying, and a caller that forgets the permit is a compile error rather than a limit that quietly does nothing.

The alternative is to park it in the spawned block with let _permit = permit;, which works and is one character away from not working: let _ = permit; is not a binding at all, so it drops the permit on the spot and the limit stops existing silently.

Choosing the number

MAX_CONNECTIONS = 128 is, again, a guess with a meaning. Multiply it by the per-connection memory (a buffer, a task, and whatever the handler allocates) to get the worst case, and check that against the memory you have. Do the same for file descriptors, because ulimit -n is often 1024 and accept failing with EMFILE is a spectacularly confusing failure mode.

The limits in this chapter compose into a pattern worth naming. Bound the number of connections, so memory is bounded. Bound the mailbox, so queueing delay is bounded. Bound the time per request, so no single one can hold a slot forever. Any one of those alone leaves a way to fall over.

Exercise

The exercise for this section is located in 06_backpressure/02_limits

Shutdown and supervision

A server that cannot be stopped cleanly is a server that loses a request on every deploy. This chapter is about the two ends of a process’s life going wrong: being asked to stop, and having something stop without being asked.

Being asked to stop

SIGTERM arrives, from Kubernetes or systemd or somebody’s Ctrl-C. What should happen is:

1. Stop accepting new connections.
2. Let the connections in flight finish, within reason.
3. Let the store task drain its mailbox.
4. Exit.

What happens by default is that main returns and the process ends, taking every task with it, mid-request. The runtime does not wait for spawned tasks and does not run their destructors.

Two tools from tokio-util do most of the work:

CancellationToken is a broadcast “please stop”. It is cheap to clone, cancelled() is a future any task can select on, and cancelling is idempotent. It is cooperative: it asks, and each task decides where it is safe to stop.

TaskTracker is a JoinSet that does not own the results. Spawn through it, then close() it and wait() for everything spawned to finish. That is the draining step, and doing it by hand with a Vec<JoinHandle> is where the bugs live.

Between them they cover both halves: cancellation ends the accept loop, and the tracker drains what is already running. Neither knows about the other, which is why the ordering is yours to get right.

Something stopping without being asked

A panicking task does not bring the process down. It unwinds, its JoinHandle starts returning Err(JoinError), and if nobody is holding that handle, nothing anywhere notices.

For a connection task that is exactly right: one client’s bad day is not everybody’s. For the store task it is a disaster, because minidb’s entire state was in it. What is left is a process that still accepts connections, still answers, and answers ERR the store is gone to every request forever. Every health check that asks “is the port open” says yes.

Rust has no supervisor tree, so somebody has to await the thing that matters and decide what its death means. Here, dying is the right answer: exit, and let whatever supervises the process do its job. mpsc::Sender::closed is how you find out, because the receiver is dropped when the store task ends, whether it returned or panicked.

The question to ask about every task

For each tokio::spawn in a codebase: who finds out if this dies, and what do they do about it?

Three answers are legitimate. Nobody needs to know, and the work is genuinely optional. Someone holds the handle and restarts it. Or its death is fatal and the process should end. What is not legitimate is not having asked, which is the default for every spawn ever written in a hurry.

Exercise

The exercise for this section is located in 07_shutdown/00_intro

Draining

serve today is a loop with no way out of it. It accepts, it spawns, it goes round again, and the only thing that ever ends it is the process ending:

#![allow(unused)]
fn main() {
loop {
    let (stream, _) = listener.accept().await?;   // nothing here is watching for a reason to stop
    // ...
}
}

Graceful shutdown is three mechanisms doing three different jobs.

Stop taking new work, by racing the accept against the token:

#![allow(unused)]
fn main() {
shutdown.cancelled().await;   // resolves once anybody has called cancel(), and stays resolved
}

accept is cancel safe, so losing that future when the token wins costs nothing: a connection that had not been accepted yet simply stays in the kernel’s backlog, and closing the listener means the client gets a connection refused and can go elsewhere.

Wait for the work in flight, by spawning through a tracker instead of through tokio::spawn:

#![allow(unused)]
fn main() {
connections.spawn(future);    // same as tokio::spawn, but counted
connections.close();          // no more will be added
connections.wait().await;     // resolves when every counted task has finished
}

close() says no more tasks will be added; wait() resolves when every task spawned through it has finished. Without the close(), wait() never returns, which is the single most common way to get this wrong.

Every await in the loop, not just the interesting one

The accept is not the only place this loop stops. Chapter 6 put a Semaphore in front of it, and at capacity the loop parks on the acquire instead, where nothing is watching the token. Cancel then, and the server carries on waiting for a permit; when one frees, select! picks between a ready cancelled() and a ready accept() at random, so a server that was told to stop accepts roughly half the clients that turn up anyway. One of the tests says so, with a limit of zero, because a server that has no permits to hand out can only leave that loop by looking at the token.

The general point is worth more than the fix: cancellation is a property of the whole loop, not of one await in it. Every point where an iteration can block is a point where a shutdown can be missed, and the only way to find them is to go through them one at a time and ask what is watching.

What “finish what they are doing” means

The connection tasks are not cancelled, but they are told. handle_connection takes the token too, and its select! gains one arm:

#![allow(unused)]
fn main() {
let line = tokio::select! {
    line = requests.next_line() => line?,
    _ = housekeeping.tick() => continue,
    _ = &mut idle_deadline => return Ok(()),
    _ = shutdown.cancelled() => return Ok(()),   // stop waiting for another request
};
}

Which await that arm cancels is the whole design. It cancels the wait for the next request, and nothing else. A request already parsed is applied and answered below the select!, untouched, so a client mid-request still gets its answer. What it does not get is another turn.

Without that arm, the unit of work is the connection, and a client that sends a request every twenty seconds keeps the drain going forever. With it, the unit of work is the request, and the drain is bounded by the slowest single request rather than by client behaviour. That is what HTTP servers do when they answer Connection: close instead of reusing a keep-alive connection.

Then bound the waiting anyway. GRACE is the constant, timeout is the tool, and after it expires whatever is left is dropped on the floor. Every await in this exercise is already bounded, so nothing should reach that deadline; the case it exists for is the one nobody bounded, such as write_all to a client that has stopped reading, where the kernel’s send buffer fills and the write never completes. Kubernetes gives a pod terminationGracePeriodSeconds, thirty by default, before SIGKILL, so your own grace period wants to sit comfortably under whatever that is set to.

Note which way round the three mechanisms go. The token stops the loop taking new connections and stops each connection taking new requests; the deadline stops the drain taking forever. A shutdown with only the first hangs on its slowest client, and one with only the last cuts off work it had already accepted.

Ordering

The order matters and follows the data:

1. Stop accepting.
2. Drain the connections. They are the only things that talk to the store.
3. Then the store task, which ends by itself once every StoreHandle has been dropped, because the channel closes when the last sender goes.

That last point is worth dwelling on. The store task’s lifetime is managed entirely by ownership of its senders. Hold a StoreHandle in a global, or in a struct that outlives the drain, and the store task will never see its channel close and shutdown will hang. When shutdown hangs, look for the handle nobody dropped.

Where the signal comes from

In production the token gets cancelled by a signal handler:

#![allow(unused)]
fn main() {
tokio::spawn(async move {
    let _ = tokio::signal::ctrl_c().await;
    shutdown.cancel();
});
}

tokio::signal also has unix::signal(SignalKind::terminate()) for SIGTERM, which is what orchestrators actually send. The exercise passes the token in from the test instead, which is the same thing with a more convenient source.

Exercise

The exercise for this section is located in 07_shutdown/01_graceful

Supervision

The store task is the one task in minidb that nothing can replace. If it dies, every handle to it answers ERR the store is gone, forever, and the server keeps accepting connections so that it can keep saying so.

A process that is up but cannot do anything is worse than one that is down. Nothing restarts it, and every liveness check that only asks whether the port is open reports success.

The accept loop you wrote in the last exercise watches two things, and the store is not one of them:

#![allow(unused)]
fn main() {
tokio::select! {
    _ = shutdown.cancelled() => /* ... */,
    accepted = listener.accept() => /* ... */,
}
}

StoreHandle now exposes the third thing it could be watching:

#![allow(unused)]
fn main() {
store.closed().await;   // mpsc::Sender::closed: resolves once the Receiver has been dropped
}

The receiver is dropped when the store task ends, whether it returned normally or panicked, so this is one await that answers “is the thing I depend on still alive”. What serve should do when it resolves is the exercise.

Dying on purpose

serve returns an error, main returns it, the process exits non-zero, and systemd or Kubernetes or a shell loop starts it again. That is supervision: not a framework, just a decision about which failures are fatal and something outside the process that notices.

Choosing to die is a real answer and usually the right one when the state that was lost lived in the task that died. minidb could not restart the store task with the data intact even if it tried, because the data was in it. After chapter 9 it could: the log on disk is what makes a restart recover rather than forget.

Restarting instead

When the dead task is stateless or its state is recoverable, supervise it by holding the handle:

#![allow(unused)]
fn main() {
loop {
    match handle.await {
        Ok(()) => break,
        Err(error) if error.is_panic() => {
            warn!(?error, "worker panicked, restarting");
            handle = tokio::spawn(worker());
        }
        Err(_) => break,   // aborted
    }
}
}

JoinError distinguishes a panic from an abort, which is the difference between “something broke” and “we asked it to stop”. Restart loops need a backoff and a give-up count, or a task that panics on startup becomes a busy loop that panics several thousand times a second.

Panics in connection tasks

Those are fine to swallow, and the semaphore permit from chapter 6 is why: it is released by Drop, so a panicking connection gives its permit back on the way out. That is the general defence. Anything that must happen when a task ends should hang off a destructor rather than off the last line of the task body, because the last line is exactly what a panic skips.

If you want a service to be loud about it, std::panic::set_hook at startup gets you one place where every panic in the process is logged, with a backtrace, before unwinding starts.

Exercise

The exercise for this section is located in 07_shutdown/02_supervision

Testing async code

Async code is hard to test for three reasons, and Tokio has an answer to each of them.

It takes time

The idle timeout is thirty seconds and no suite can wait. Pause the clock:

#![allow(unused)]
fn main() {
#[tokio::test(start_paused = true)]
async fn a_client_that_says_nothing_is_hung_up_on() {
}

With time paused, the runtime auto-advances: whenever every task is blocked on a timer, it jumps straight to the earliest deadline. Thirty seconds pass in microseconds, Instant::now() agrees they did, and the assertion can be == rather than a tolerance window.

tokio::time::advance(duration) moves the clock by hand when you want the deadline to arrive at a particular point in your test rather than as soon as possible.

The one thing that surprises everybody: a paused clock does not notice work. Time only moves when the runtime decides it should, so a loop burning two million iterations takes exactly zero nanoseconds as far as Instant is concerned. That is not a limitation, it is the property that makes these tests deterministic. It also means you cannot use a paused clock to measure anything, only to test behaviour that depends on time passing.

Anything that waits by not awaiting a timer is invisible to all of this: std::thread::sleep inside spawn_blocking sleeps for real. Which is one more reason to make waiting explicit.

It needs a peer

For a server, the peer is a socket. Two ways to avoid the pain:

Port zero. TcpListener::bind("127.0.0.1:0") and then local_addr() gives a hermetic test that can run in parallel with itself.

No socket at all. tokio::io::duplex(1024) returns two connected in-memory pipes implementing AsyncRead + AsyncWrite. Hand one half to handle_connection and keep the other:

#![allow(unused)]
fn main() {
let (client, server) = tokio::io::duplex(1024);
tokio::spawn(async move { handle_connection(server, &store, IDLE).await });
}

This is why handle_connection is generic over its stream, and it is worth noticing that the generic was not added for reuse: it was added for testability, and it is the reason the cancel-safety test in chapter 5 could send twelve bytes, wait, and send the rest.

It is concurrent

Which is to say the failure happens once in fifty runs on your machine and every time on somebody else’s. Most of the fix is in the design rather than the test: a single task owning the store is testable in a way that a lock held across an await is not, because there is one place where ordering is decided.

What helps in the tests themselves:

• Assert on properties, not on schedules. “Some were shed and some were accepted” survives a Tokio upgrade; “exactly three were shed” does not.
• Use #[tokio::test]’s default current-thread runtime when you want deterministic ordering, and #[tokio::test(flavor = "multi_thread")] when the thing you are testing is that it works when it is not.
• Bound anything that could hang with a timeout, so a broken implementation fails with a message instead of stopping CI.
• yield_now().await in a loop is a blunt but honest way to let spawned tasks reach their next await before you assert.

Exercise

The exercise for this section is located in 08_testing/00_intro

Retrying, and testing that it waited

The client connects once, and if nobody is listening it gives up:

#![allow(unused)]
fn main() {
let stream = TcpStream::connect(&addr).await?;   // one go, and then out
}

Every real client retries, and every retry that does not back off turns one restarting server into a thundering herd. with_backoff is the wrapper that fixes it, and its signature is the interesting part:

#![allow(unused)]
fn main() {
pub async fn with_backoff<O, F, T, E>(attempts: u32, base: Duration, operation: O) -> Result<T, E>
where
    O: FnMut() -> F,
    F: Future<Output = Result<T, E>>,
}

The shape underneath is a loop around three outcomes. A call that succeeded returns its value. A call that failed with attempts still to come waits and goes round again. A call that failed on the last attempt returns that error, and the count of attempts made is what tells those last two apart. The wait doubles by raising two to the number of failures so far, and it lives on the failure path only, so nothing is waited for after the final attempt.

Why the argument is a closure

Not FnMut() -> Result<T, E>, and not a single future.

A future runs once. Awaiting it consumes it, and there is no way to rewind it, so retrying means asking for a new one each time. That is what FnMut() -> F expresses: a thing that makes futures.

This shape turns up all over async Rust, in retry helpers, connection pools, anything that supervises, and it is worth being able to write from memory. Note also that F is one type parameter, so every call must produce the same future type, which an async block or a direct call satisfies and a match returning two different futures does not. Box::pin is the escape hatch.

The tests are the chapter

Four tests assert the exact elapsed time of a retry sequence, up to a second and a half of it, and the suite finishes in microseconds:

#![allow(unused)]
fn main() {
assert_eq!(started.elapsed(), BASE_DELAY * 15);   // 100 + 200 + 400 + 800
}

No tolerance window, no sleep in the test, no flake on a loaded laptop. That is the deal a paused clock offers: code that waits by awaiting a timer is testable to the millisecond.

Note what the tests pin down beyond the total. That the first attempt is immediate. That the wait doubles rather than being constant. That there is no sleep after the last attempt, which is the detail everybody’s first implementation gets wrong and nobody’s first test catches, because a stray hundred milliseconds at the end of a retry loop is invisible unless you are measuring.

What is missing from this backoff

Two things, deliberately, and both are worth adding in production code:

Jitter. A thousand clients that all fail at the same moment and all back off by exactly 100ms retry at exactly the same moment. Randomising the delay (base * 2^n * random(0.5..1.5), or the full-jitter variant that picks uniformly from the whole interval) spreads the retry storm out.

A cap. Doubling forever reaches absurd delays quickly, and 2u32.pow(attempt - 1) overflows at 33 attempts. Real backoff clamps to a maximum, usually tens of seconds.

Neither is in the exercise, because both make the arithmetic in the tests less obvious, which is a trade worth understanding: this implementation is exactly testable because it is exactly predictable, and adding jitter means the test has to assert on a range. That is a fair price, but it should be a deliberate one.

Exercise

The exercise for this section is located in 08_testing/01_time

Seeing inside

minidb handles hundreds of connections at once and says nothing about any of them. When one client in a hundred gets ERR busy, there is no way to find out which, when, or what it had asked for.

println! does not scale to this. Interleaved output from a hundred tasks is a soup, because a line of text carries no record of which piece of work produced it.

Spans and events

tracing splits it in two. A span is a period of work with a beginning and an end. An event is a moment. Every event records which spans it happened inside, so the context travels with the data instead of being copied into every message.

#![allow(unused)]
fn main() {
#[instrument]
async fn upload(id: u64) {
    // everything in here, including anything it awaits, happens inside a span named "upload"
}
}

#[instrument] opens a span for the whole function, including across every .await in it, which is the part that matters: the span is attached to the future, so it is entered and exited every time the task is polled and the context survives suspension. This is why tracing and not log.

Two details from the attribute:

• skip_all. #[instrument] records every argument as a field and therefore requires them all to be Debug. A generic stream and a StoreHandle are not. skip_all records none of them, and fields(...) adds back the ones worth having.
• name = "connection". A span name is something an operator reads. A function name is an implementation detail that will be refactored next month.

Fields, not sentences

#![allow(unused)]
fn main() {
info!(bytes = %written, peer = %addr, "wrote");   // yes
info!("wrote {written} bytes to {addr}");         // no
}

Both print about the same thing. Only the first can be indexed, so that a collector can answer “how many ERR busy in the last hour” without anybody writing a regex. % records the Display output, ? records Debug, and a bare name records the value.

Levels are part of the same discipline. A client sending nonsense is a warn! at most: it is not your server malfunctioning, and it must not be able to fill your error budget by typing badly.

Testing instrumentation

The exercise asserts on spans and fields directly, with a Layer of its own that collects events into a Vec rather than printing them:

#![allow(unused)]
fn main() {
let captured = Captured::default();
let _guard = tracing::subscriber::set_default(Registry::default().with(captured.clone()));
}

That is the part worth taking home. Instrumentation is structured data, so it can be tested like data, and a log line your alerting depends on deserves a test as much as any other behaviour does. set_default scopes the subscriber to the current thread, which is exactly right under #[tokio::test]’s single-threaded runtime.

In the binaries

#![allow(unused)]
fn main() {
tracing_subscriber::fmt::init();
}

one line in main, and RUST_LOG=info cargo run prints the events. In production the same events go to a JSON layer, or to OpenTelemetry via tracing-opentelemetry, where the spans become distributed traces. None of the instrumentation changes; only the subscriber does.

Exercise

The exercise for this section is located in 08_testing/02_tracing

Surviving a restart

minidb keeps everything in a HashMap, so a restart loses the lot. The fix is the oldest idea in databases: before you change anything, write down what you are about to do, somewhere that outlives the process.

That is a write-ahead log, and the two words are the whole idea. Write it ahead of the change, because a log written afterwards is missing exactly the records you needed.

The format is already here

Every mutating request is a line of the wire protocol:

SET users alice hello
DEL users alice

So the log is a transcript of what clients asked for, and replaying it is running those requests again in order. Request has parse and Display and a round-trip test from chapter 3, so the log writer and the log reader were finished before this chapter started.

That is not a trick to save time in a workshop. Reusing the wire format as the log format is what gives you a log you can read with cat, and it means one round-trip test covers both.

The thing to be careful about is that the log records requests, not results. SET users alice hello replays to the same state every time. INCR users counter would not, and a log of non-deterministic operations replays into a different database than the one you had. If a command can produce a different result on a different day, log its effect rather than the command.

write_all is not durability

#![allow(unused)]
fn main() {
self.file.write_all(format!("{request}\n").as_bytes()).await
}

hands the bytes to tokio::fs, which hands them to the operating system, which puts them in a cache and says it is done. A process crash is survivable at that point. A power cut is not.

sync_all is the call that waits for the disk, and it is expensive, which is why the next two exercises are about when to call it rather than whether.

There are two buffers in the way, and both have to be emptied:

#![allow(unused)]
fn main() {
pub async fn sync(&mut self) -> io::Result<()> {
    self.file.flush().await?;      // tokio's own buffer
    self.file.sync_all().await     // the operating system's
}
}

Tokio’s file I/O is not async

There is no portable way to await a disk, so tokio::fs wraps the blocking calls in spawn_blocking. Chapter 2, in other words, with the trip to the blocking pool already written for you.

Two consequences follow. Every append costs a trip to that pool, which is another reason to do more per trip. And write_all returns before the write has been attempted, so a disk that refuses it says nothing until the buffer is flushed. The test for a log that cannot be written is where that shows up, and it is the reason the durability check in the next exercise has to look at what sync returned.

Where this chapter goes

Write the record before applying the change. Batch the syncs so a busy server does not make one trip to the disk per request. Replay the log on startup so the restart is invisible to whoever reconnects.

Exercise

The exercise for this section is located in 09_wal/00_intro

Write ahead

The store task now owns a Wal as well as a Store, and run logs before it applies:

#![allow(unused)]
fn main() {
if let Err(error) = log(&mut wal, &request).await {
    let _ = reply.send(Response::Error(format!("not logged: {error}")));
    continue;
}

let _ = reply.send(apply(request, &mut store));
}

Two decisions live in log.

What gets logged

A GET changes nothing, so writing it down would cost a disk sync to record that nothing happened. SET and DEL are the log, and matches!(request, Request::Get { .. }) is how you say so.

Wal gives you two calls, and the difference between them is the whole chapter:

#![allow(unused)]
fn main() {
wal.append(&request).await;   // -> io::Result<()>, into a buffer, not onto a disk
wal.sync().await;             // -> io::Result<()>, and back only once the disk agrees
}

When it is safe to say yes

append hands the bytes to a buffer, so a reply sent after the append and before the sync is a promise you have not kept. The reply goes out after the sync, and only if the sync said it worked.

The exercise’s second test is the one that pins this down: a store whose log cannot be written must refuse the change, not apply it and hope. A server that acknowledges a write it could not record is the failure mode this entire chapter exists to prevent, and it is invisible until the day the disk is full.

The asymmetry

Say it in both directions, because only one of them is a problem.

The log may contain changes the store never applied, because the process can die between the sync and the apply. That is fine. Replaying a change that already happened sets the same key to the same value, which is why replay wants idempotent records.

A store that is ahead of its log is data loss, and no amount of replaying fixes it. Every rule in this chapter follows from that one asymmetry.

What this costs

One sync per request, which is the slowest thing this server now does. A spinning disk does on the order of a hundred of those a second; an SSD does more, and still fewer than the store task could apply. From here on, durability is the bottleneck, and the next exercise is about the standard way of making it a smaller one.

Exercise

The exercise for this section is located in 09_wal/01_append

Group commit

One sync per request is correct and slow. A sync is a round trip to a physical device, and tokio::fs runs it on the blocking pool, so a hundred requests a second is a hundred trips to the disk and a hundred trips to the pool, whether or not those requests arrived together.

What the last exercise left you with pays that cost once per request, however many are waiting:

#![allow(unused)]
fn main() {
wal.append(request).await?;
wal.sync().await          // one trip to the disk, for one change
}

They usually did arrive together. That is what a mailbox is.

#![allow(unused)]
fn main() {
let mut batch = Vec::with_capacity(BATCH);

while inbox.recv_many(&mut batch, BATCH).await > 0 {
    if let Err(error) = commit(&mut wal, &batch).await {
        // refuse the whole batch
    }

    for Command { request, reply } in batch.drain(..) {
        let _ = reply.send(apply(request, &mut store));
    }
}
}

recv_many takes everything that is waiting, up to a limit, in one call. Sixteen requests that arrived while the last sync was in flight become one Vec rather than sixteen turns of the loop. That loop is written for you; commit is not. It gets the whole batch, and the same two calls as before, and has to decide how many times to make each of them.

Two things fall out of that. Appending is per record and syncing is not, so the count of append calls and the count of sync calls are different numbers. And a batch that was all reads syncs nothing at all, which is the same decision as the last exercise applied to a set, so commit has to know whether the batch changed anything before it decides to sync.

Why this is not cheating

Durability is only promised to a client that has been answered, and nothing in the batch is answered until the sync returns. The sixteenth client waits no longer than it would have; the first one waits slightly longer than it would have. Every one of them gets exactly the same promise as before, and the disk did one trip instead of sixteen.

This is group commit, and every database you have used does it. Postgres calls it commit delay, and will even wait a moment before syncing to let more transactions join the batch, trading latency for throughput on purpose.

The idea generalises well past disks: when work has a fixed cost per trip, the thing to batch is the trip. Network round trips, syscalls, lock acquisitions, writes to a metrics backend, all the same shape.

Testing a performance property

The test spawns sixteen requests at once and counts the syncs:

#![allow(unused)]
fn main() {
assert!(syncs <= 4, "sixteen requests that arrived together cost {syncs} syncs");
}

Counting a side effect is what makes this testable at all. A wall-clock assertion would be a flake waiting for a slower machine; the number of times sync was called is exact, and it is what you actually mean by “batched”.

The bound is <= 4 rather than == 1 deliberately. Batching sixteen into one is what happens today on a current-thread runtime, and pinning that exactly would make the test a hostage to the scheduler’s arrival timing. Four is comfortably below sixteen, and no implementation that syncs per request can sneak past it.

Exercise

The exercise for this section is located in 09_wal/02_group_commit

Replay

The log has been correct since the start of this chapter and has never once been read. minidb still starts every time with Store::new() and no memory of anything, and reading the log back is the last thing it needs.

Replaying is running the requests again, in order, against an empty store. Everything it takes already exists: File::open and the same BufReader::lines that reads the wire, Request::parse from chapter 3, and apply. This function is glue, and that is the point of having chosen the wire format as the log format.

#![allow(unused)]
fn main() {
File::open(path).await;      // -> io::Result<File>, and the error kind matters here
error.kind();                // -> ErrorKind, which has a NotFound worth treating separately
io::Error::other(message);   // for turning a parse failure into something this can return
}

Two cases worth deciding on purpose

No log at all. ErrorKind::NotFound is not a failure, it is what a first start looks like. Any other error from File::open is a real problem and belongs to the caller: a log that exists and cannot be opened is not the same as no log, and starting empty in that case would silently discard a database.

A line that does not parse. Refuse to start. A server that skips records it cannot read comes up quietly holding a database that is missing writes it acknowledged, and nobody finds out until much later.

There is a more sophisticated version of that second rule, and it is what real systems do. A crash mid-write leaves a torn last record, so the convention is to accept a truncated final line, discard it, and refuse anything malformed in the middle. Doing it properly means a checksum per record, so that a record which is complete but corrupt is detected rather than replayed.

Proving it by hand

cargo run                              # terminal one
cargo run --bin client                 # terminal two
SET users alice hello

Ctrl-C the server, start it again, ask for the key back, and it is there. The log is a text file in the working directory; cat minidb.wal shows exactly what was recorded.

Where to go next

What you have is a real write-ahead log with a real weakness: it grows forever, and a restart takes as long as the entire history of the database.

• Checkpointing. Periodically write the current state out in full, then truncate the log up to that point. Restart cost becomes the size of the data rather than the size of its history.
• Segments. One file per span of the log, so old segments can be deleted or archived without rewriting anything.
• Checksums. Per record, so corruption is detected rather than replayed.
• fdatasync and O_DIRECT. The next layer of the durability story, and the point where the answers become filesystem-specific.
• Group commit with a delay. Wait a moment before syncing so more requests can join the batch, trading a little latency for throughput.

Those are the next things to build, and they are all yours. The book stays where you left it.

Exercise

The exercise for this section is located in 09_wal/03_replay