Workshop 3: Async I/O

This workshop builds five programs with the async I/O tools from the previous lessons. Each project starts with a problem statement and a list of requirements, then grows through stages. Every stage is a complete program: copy it into src/main.rs of a project that depends on tokio with the full feature, and it compiles and runs.

Project	Program	Main tools
10	TCP echo server and client	`TcpListener`, `spawn` per connection, `read`, `write_all`, `into_split`, `shutdown`
11	Key-value server with a text protocol	`BufReader::lines`, `Arc<Mutex<HashMap>>`, `split`
12	Chat room server	`broadcast`, `select!`, `lines`, `into_split`
13	Minimal HTTP/1.1 server	manual parsing, routing, `AsyncWriteExt`, `io::split`, `duplex`
14	Process runner and file follower	`process::Command`, `fs`, `timeout`, `interval`, piped output

Every server binds 127.0.0.1:0, so the operating system picks a free port, and the program drives itself with a client running in the same process. The programs print no addresses, ports, paths, or timings. Where two events could happen in either order, the client script waits for the first before starting the second.

Project 10: TCP echo server and client

An echo server sends back every byte it receives. It is the smallest program that has the structure of every TCP server: bind, accept, handle each connection independently, and deal with the end of the stream.

Requirements:

The server binds 127.0.0.1:0 and accepts connections in a loop.
Each connection runs in its own task, so a slow client does not block the others.
The server writes back exactly the bytes it read, in order, for any message size.
A connection ends when the client closes its sending side.
The server counts connections and echoed bytes, and stops on request.

Stage 1: one connection

The server accepts a single connection and echoes until the client closes. read returns 0 at end of stream. The client knows the reply length, so it uses read_exact.

Rust

use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let addr = listener.local_addr()?;

    let server = tokio::spawn(async move {
        let (mut socket, _peer) = listener.accept().await?;
        let mut buf = [0u8; 1024];
        loop {
            let n = socket.read(&mut buf).await?;
            if n == 0 {
                return Ok::<_, std::io::Error>(());
            }
            socket.write_all(&buf[..n]).await?;
        }
    });

    let mut client = TcpStream::connect(addr).await?;
    client.write_all(b"hello").await?;
    let mut reply = [0u8; 5];
    client.read_exact(&mut reply).await?;
    println!("echoed: {}", String::from_utf8_lossy(&reply));

    drop(client);
    server.await.expect("server task panicked")?;
    println!("server finished");
    Ok(())
}

Output

echoed: hello
server finished

Dropping the client closes the connection. The server's next read returns 0, the loop returns, and the server task ends.

Stage 2: an accept loop and many clients

The handler moves into a function, and the server accepts forever. Each accepted socket goes to a new task. The client helper sends a message, closes its sending side with shutdown, and reads until the server closes. Four clients run at once, and awaiting their handles in order keeps the printed order fixed.

Rust

use std::net::SocketAddr;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};

async fn echo(mut socket: TcpStream) -> std::io::Result<usize> {
    let mut buf = [0u8; 1024];
    let mut total = 0;
    loop {
        let n = socket.read(&mut buf).await?;
        if n == 0 {
            return Ok(total);
        }
        socket.write_all(&buf[..n]).await?;
        total += n;
    }
}

async fn serve(listener: TcpListener) {
    loop {
        match listener.accept().await {
            Ok((socket, _peer)) => {
                tokio::spawn(async move {
                    if let Err(e) = echo(socket).await {
                        eprintln!("connection error: {e}");
                    }
                });
            }
            Err(e) => eprintln!("accept error: {e}"),
        }
    }
}

async fn echo_once(addr: SocketAddr, message: &str) -> std::io::Result<String> {
    let mut client = TcpStream::connect(addr).await?;
    client.write_all(message.as_bytes()).await?;
    client.shutdown().await?;
    let mut reply = String::new();
    client.read_to_string(&mut reply).await?;
    Ok(reply)
}

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let addr = listener.local_addr()?;
    let server = tokio::spawn(serve(listener));

    let mut clients = Vec::new();
    for i in 0..4 {
        clients.push(tokio::spawn(async move {
            echo_once(addr, &format!("message from client {i}")).await
        }));
    }
    for (i, client) in clients.into_iter().enumerate() {
        let reply = client.await.expect("client panicked")?;
        println!("client {i} got: {reply}");
    }

    server.abort();
    Ok(())
}

Output

client 0 got: message from client 0
client 1 got: message from client 1
client 2 got: message from client 2
client 3 got: message from client 3

shutdown on the client closes only its sending direction. The server sees end of stream, finishes its loop, and drops the socket, which makes the client's read_to_string return. The server task loops forever, so the program ends it with abort.

Stage 3: statistics, large messages, and a clean stop

The final program adds three things:

A Stats value behind Arc<Mutex<_>> counts connections and bytes. The lock is held only for the increment, never across an .await.
A large-message client splits its socket with into_split. One task writes the payload while the main task reads the echo. A single task that wrote everything before reading could stall once the operating system's buffers fill, because the server would be blocked writing echoes that nobody reads.
The accept loop uses select! to stop when a oneshot message arrives.

Rust

use std::net::SocketAddr;
use std::sync::{Arc, Mutex};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::oneshot;

#[derive(Default)]
struct Stats {
    connections: u64,
    bytes: u64,
}

type Shared = Arc<Mutex<Stats>>;

async fn echo(mut socket: TcpStream, stats: Shared) -> std::io::Result<()> {
    stats.lock().unwrap().connections += 1;
    let mut buf = [0u8; 4096];
    loop {
        let n = socket.read(&mut buf).await?;
        if n == 0 {
            return Ok(());
        }
        stats.lock().unwrap().bytes += n as u64;
        socket.write_all(&buf[..n]).await?;
    }
}

async fn serve(listener: TcpListener, stats: Shared, mut stop: oneshot::Receiver<()>) {
    loop {
        tokio::select! {
            accepted = listener.accept() => match accepted {
                Ok((socket, _peer)) => {
                    let stats = Arc::clone(&stats);
                    tokio::spawn(async move {
                        if let Err(e) = echo(socket, stats).await {
                            eprintln!("connection error: {e}");
                        }
                    });
                }
                Err(e) => eprintln!("accept error: {e}"),
            },
            _ = &mut stop => break,
        }
    }
}

async fn echo_once(addr: SocketAddr, message: &str) -> std::io::Result<String> {
    let mut client = TcpStream::connect(addr).await?;
    client.write_all(message.as_bytes()).await?;
    client.shutdown().await?;
    let mut reply = String::new();
    client.read_to_string(&mut reply).await?;
    Ok(reply)
}

async fn echo_large(addr: SocketAddr, payload: Vec<u8>) -> std::io::Result<Vec<u8>> {
    let (mut read_half, mut write_half) = TcpStream::connect(addr).await?.into_split();
    let to_send = payload;
    let writer = tokio::spawn(async move {
        write_half.write_all(&to_send).await?;
        write_half.shutdown().await
    });
    let mut echoed = Vec::new();
    read_half.read_to_end(&mut echoed).await?;
    writer.await.expect("writer panicked")?;
    Ok(echoed)
}

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let addr = listener.local_addr()?;
    let stats: Shared = Arc::default();
    let (stop_tx, stop_rx) = oneshot::channel();
    let server = tokio::spawn(serve(listener, Arc::clone(&stats), stop_rx));

    let mut clients = Vec::new();
    for i in 0..4 {
        clients.push(tokio::spawn(async move {
            echo_once(addr, &format!("message from client {i}")).await
        }));
    }
    for (i, client) in clients.into_iter().enumerate() {
        let reply = client.await.expect("client panicked")?;
        println!("client {i} got {} bytes: {reply}", reply.len());
    }

    let payload: Vec<u8> = (0..300_000u32).map(|i| (i % 251) as u8).collect();
    let echoed = echo_large(addr, payload.clone()).await?;
    println!("large payload of {} bytes intact: {}", payload.len(), echoed == payload);

    {
        let stats = stats.lock().unwrap();
        println!("connections: {}, bytes echoed: {}", stats.connections, stats.bytes);
    }

    stop_tx.send(()).expect("server still running");
    server.await.expect("server panicked");
    println!("server stopped");
    Ok(())
}

Output

client 0 got 21 bytes: message from client 0
client 1 got 21 bytes: message from client 1
client 2 got 21 bytes: message from client 2
client 3 got 21 bytes: message from client 3
large payload of 300000 bytes intact: true
connections: 5, bytes echoed: 300084
server stopped

The counters are exact because each handler increments them before it writes the echo, and a client only finishes after it has read the whole echo. The large payload arrives in many read calls of at most 4096 bytes, and the echo is still byte-for-byte identical because TCP delivers bytes in order.

Part	Role
`serve`	accepts connections, spawns one task per connection, stops on the `oneshot` message
`echo`	reads chunks and writes them back until end of stream, updates `Stats`
`Stats` behind `Arc<Mutex<_>>`	shared counters, locked only for a single increment
`echo_once`	client for short messages: write, `shutdown`, read to the end
`echo_large`	client for large messages: `into_split` so writing and reading overlap

Extensions:

Close connections that send nothing for a period. Wrap the read call in timeout and count timed-out connections in Stats.
Limit the server to 100 simultaneous connections with a Semaphore. Acquire an owned permit before spawn and move it into the task.
Add a bytes_by_connection list to Stats and print the sizes in connection order.

Project 11: Key-value server with a text protocol

A key-value server stores values by key and answers text commands, one per line. Because the protocol is plain text, a person can type it into a terminal. Every command needs a defined reply, including the error replies.

Requirements:

Commands are SET key value, GET key, DEL key, COUNT, and QUIT. Later stages add INCR key and KEYS.
Each command produces exactly one reply line: OK, VALUE v, DELETED, NOT_FOUND, COUNT n, BYE, or ERR message.
A value may contain spaces. A key may not.
The store is shared by all connections and never held locked across an .await.
A client may send many commands without waiting for replies, and receives the replies in order.

Stage 1: parse and execute, with no networking

The protocol logic is plain functions on strings and a HashMap. Testing it without sockets keeps the first stage small and shows the exact set of replies.

Rust

use std::collections::HashMap;

#[derive(Debug, PartialEq)]
enum Command {
    Set(String, String),
    Get(String),
    Del(String),
    Count,
    Quit,
}

fn parse(line: &str) -> Result<Command, String> {
    let line = line.trim();
    let mut parts = line.splitn(3, ' ');
    let name = parts.next().unwrap_or("").to_ascii_uppercase();
    let (first, second) = (parts.next(), parts.next());
    match (name.as_str(), first, second) {
        ("SET", Some(key), Some(value)) => Ok(Command::Set(key.to_string(), value.to_string())),
        ("GET", Some(key), None) => Ok(Command::Get(key.to_string())),
        ("DEL", Some(key), None) => Ok(Command::Del(key.to_string())),
        ("COUNT", None, None) => Ok(Command::Count),
        ("QUIT", None, None) => Ok(Command::Quit),
        ("", None, None) => Err("empty command".to_string()),
        _ => Err(format!("bad command '{line}'")),
    }
}

fn execute(command: Command, store: &mut HashMap<String, String>) -> String {
    match command {
        Command::Set(key, value) => {
            store.insert(key, value);
            "OK".to_string()
        }
        Command::Get(key) => match store.get(&key) {
            Some(value) => format!("VALUE {value}"),
            None => "NOT_FOUND".to_string(),
        },
        Command::Del(key) => {
            if store.remove(&key).is_some() {
                "DELETED".to_string()
            } else {
                "NOT_FOUND".to_string()
            }
        }
        Command::Count => format!("COUNT {}", store.len()),
        Command::Quit => "BYE".to_string(),
    }
}

fn main() {
    let script = [
        "SET city Lisbon",
        "SET motto keep it simple",
        "GET motto",
        "get city",
        "GET missing",
        "COUNT",
        "DEL city",
        "DEL city",
        "SET lonely",
        "",
        "FLY away",
        "QUIT",
    ];

    let mut store = HashMap::new();
    for line in script {
        let reply = match parse(line) {
            Ok(command) => execute(command, &mut store),
            Err(message) => format!("ERR {message}"),
        };
        println!("{line:<24} -> {reply}");
    }
}

Output

SET city Lisbon          -> OK
SET motto keep it simple -> OK
GET motto                -> VALUE keep it simple
get city                 -> VALUE Lisbon
GET missing              -> NOT_FOUND
COUNT                    -> COUNT 2
DEL city                 -> DELETED
DEL city                 -> NOT_FOUND
SET lonely               -> ERR bad command 'SET lonely'
                         -> ERR empty command
FLY away                 -> ERR bad command 'FLY away'
QUIT                     -> BYE

The command name is case-insensitive, so get works. splitn(3, ' ') splits into at most three parts, so everything after the key stays together as the value. Error text is not part of the command enum: parse returns Err(String), and the caller adds the ERR prefix.

Stage 2: serve the protocol over TCP

The store becomes Arc<Mutex<HashMap<_, _>>>. Each connection task reads lines with BufReader::lines and writes replies. stream.split() borrows the stream as a reader half and a writer half, so the reader can sit inside BufReader while the writer stays usable. The lock is taken inside a small block, so the guard is gone before the next .await.

Rust

use std::collections::HashMap;
use std::net::SocketAddr;
use std::sync::{Arc, Mutex};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::{TcpListener, TcpStream};

type Store = Arc<Mutex<HashMap<String, String>>>;

#[derive(Debug, PartialEq)]
enum Command {
    Set(String, String),
    Get(String),
    Del(String),
    Count,
    Quit,
}

fn parse(line: &str) -> Result<Command, String> {
    let line = line.trim();
    let mut parts = line.splitn(3, ' ');
    let name = parts.next().unwrap_or("").to_ascii_uppercase();
    let (first, second) = (parts.next(), parts.next());
    match (name.as_str(), first, second) {
        ("SET", Some(key), Some(value)) => Ok(Command::Set(key.to_string(), value.to_string())),
        ("GET", Some(key), None) => Ok(Command::Get(key.to_string())),
        ("DEL", Some(key), None) => Ok(Command::Del(key.to_string())),
        ("COUNT", None, None) => Ok(Command::Count),
        ("QUIT", None, None) => Ok(Command::Quit),
        ("", None, None) => Err("empty command".to_string()),
        _ => Err(format!("bad command '{line}'")),
    }
}

fn execute(command: Command, store: &mut HashMap<String, String>) -> String {
    match command {
        Command::Set(key, value) => {
            store.insert(key, value);
            "OK".to_string()
        }
        Command::Get(key) => match store.get(&key) {
            Some(value) => format!("VALUE {value}"),
            None => "NOT_FOUND".to_string(),
        },
        Command::Del(key) => {
            if store.remove(&key).is_some() {
                "DELETED".to_string()
            } else {
                "NOT_FOUND".to_string()
            }
        }
        Command::Count => format!("COUNT {}", store.len()),
        Command::Quit => "BYE".to_string(),
    }
}

async fn handle(mut stream: TcpStream, store: Store) -> std::io::Result<()> {
    let (reader, mut writer) = stream.split();
    let mut lines = BufReader::new(reader).lines();
    while let Some(line) = lines.next_line().await? {
        let (reply, quit) = match parse(&line) {
            Ok(command) => {
                let quit = command == Command::Quit;
                let reply = execute(command, &mut store.lock().unwrap());
                (reply, quit)
            }
            Err(message) => (format!("ERR {message}"), false),
        };
        writer.write_all(format!("{reply}\n").as_bytes()).await?;
        if quit {
            break;
        }
    }
    Ok(())
}

async fn serve(listener: TcpListener, store: Store) {
    loop {
        let Ok((stream, _peer)) = listener.accept().await else {
            continue;
        };
        let store = Arc::clone(&store);
        tokio::spawn(async move {
            if let Err(e) = handle(stream, store).await {
                eprintln!("connection error: {e}");
            }
        });
    }
}

async fn session(addr: SocketAddr, script: &[&str]) -> std::io::Result<Vec<String>> {
    let mut stream = TcpStream::connect(addr).await?;
    let mut request = String::new();
    for line in script {
        request.push_str(line);
        request.push('\n');
    }
    stream.write_all(request.as_bytes()).await?;

    let mut replies = Vec::new();
    let mut lines = BufReader::new(stream).lines();
    while let Some(line) = lines.next_line().await? {
        replies.push(line);
    }
    Ok(replies)
}

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let addr = listener.local_addr()?;
    let store: Store = Arc::default();
    let server = tokio::spawn(serve(listener, store));

    let first = ["SET color blue", "GET color", "COUNT", "QUIT"];
    for (command, reply) in first.iter().zip(session(addr, &first).await?) {
        println!("client 1: {command:<16} -> {reply}");
    }

    let second = ["GET color", "DEL color", "GET color", "QUIT"];
    for (command, reply) in second.iter().zip(session(addr, &second).await?) {
        println!("client 2: {command:<16} -> {reply}");
    }

    server.abort();
    Ok(())
}

Output

client 1: SET color blue   -> OK
client 1: GET color        -> VALUE blue
client 1: COUNT            -> COUNT 1
client 1: QUIT             -> BYE
client 2: GET color        -> VALUE blue
client 2: DEL color        -> DELETED
client 2: GET color        -> NOT_FOUND
client 2: QUIT             -> BYE

The first client sends four commands in one write and reads four replies. The second client is a separate connection, and it sees the value the first one stored: the store is shared. The client waits for the server to close the connection, which happens after QUIT, so it reads every reply before it stops.

Stage 3: INCR, KEYS, and concurrent clients

INCR key adds 1 to an integer value and creates the key at 1 when it is missing. The read, the addition, and the write happen while one lock guard is held, so two clients incrementing at the same time cannot lose an update. KEYS lists keys in sorted order. Ten clients each send 20 INCR commands at once, and the final value is checked.

Rust

use std::collections::HashMap;
use std::net::SocketAddr;
use std::sync::{Arc, Mutex};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::{TcpListener, TcpStream};

type Store = Arc<Mutex<HashMap<String, String>>>;

#[derive(Debug, PartialEq)]
enum Command {
    Set(String, String),
    Get(String),
    Del(String),
    Incr(String),
    Keys,
    Count,
    Quit,
}

fn parse(line: &str) -> Result<Command, String> {
    let line = line.trim();
    let mut parts = line.splitn(3, ' ');
    let name = parts.next().unwrap_or("").to_ascii_uppercase();
    let (first, second) = (parts.next(), parts.next());
    match (name.as_str(), first, second) {
        ("SET", Some(key), Some(value)) => Ok(Command::Set(key.to_string(), value.to_string())),
        ("GET", Some(key), None) => Ok(Command::Get(key.to_string())),
        ("DEL", Some(key), None) => Ok(Command::Del(key.to_string())),
        ("INCR", Some(key), None) => Ok(Command::Incr(key.to_string())),
        ("KEYS", None, None) => Ok(Command::Keys),
        ("COUNT", None, None) => Ok(Command::Count),
        ("QUIT", None, None) => Ok(Command::Quit),
        ("", None, None) => Err("empty command".to_string()),
        _ => Err(format!("bad command '{line}'")),
    }
}

fn execute(command: Command, store: &mut HashMap<String, String>) -> String {
    match command {
        Command::Set(key, value) => {
            store.insert(key, value);
            "OK".to_string()
        }
        Command::Get(key) => match store.get(&key) {
            Some(value) => format!("VALUE {value}"),
            None => "NOT_FOUND".to_string(),
        },
        Command::Del(key) => {
            if store.remove(&key).is_some() {
                "DELETED".to_string()
            } else {
                "NOT_FOUND".to_string()
            }
        }
        Command::Incr(key) => {
            let current = match store.get(&key) {
                Some(text) => text.parse::<i64>().map_err(|_| ()),
                None => Ok(0),
            };
            match current {
                Ok(n) => {
                    store.insert(key, (n + 1).to_string());
                    format!("VALUE {}", n + 1)
                }
                Err(()) => "ERR value is not an integer".to_string(),
            }
        }
        Command::Keys => {
            let mut keys: Vec<&str> = store.keys().map(String::as_str).collect();
            keys.sort();
            format!("KEYS {}", keys.join(" ")).trim_end().to_string()
        }
        Command::Count => format!("COUNT {}", store.len()),
        Command::Quit => "BYE".to_string(),
    }
}

async fn handle(mut stream: TcpStream, store: Store) -> std::io::Result<()> {
    let (reader, mut writer) = stream.split();
    let mut lines = BufReader::new(reader).lines();
    while let Some(line) = lines.next_line().await? {
        let (reply, quit) = match parse(&line) {
            Ok(command) => {
                let quit = command == Command::Quit;
                let reply = execute(command, &mut store.lock().unwrap());
                (reply, quit)
            }
            Err(message) => (format!("ERR {message}"), false),
        };
        writer.write_all(format!("{reply}\n").as_bytes()).await?;
        if quit {
            break;
        }
    }
    Ok(())
}

async fn serve(listener: TcpListener, store: Store) {
    loop {
        let Ok((stream, _peer)) = listener.accept().await else {
            continue;
        };
        let store = Arc::clone(&store);
        tokio::spawn(async move {
            if let Err(e) = handle(stream, store).await {
                eprintln!("connection error: {e}");
            }
        });
    }
}

async fn session(addr: SocketAddr, script: &[String]) -> std::io::Result<Vec<String>> {
    let mut stream = TcpStream::connect(addr).await?;
    let mut request = String::new();
    for line in script {
        request.push_str(line);
        request.push('\n');
    }
    stream.write_all(request.as_bytes()).await?;

    let mut replies = Vec::new();
    let mut lines = BufReader::new(stream).lines();
    while let Some(line) = lines.next_line().await? {
        replies.push(line);
    }
    Ok(replies)
}

fn script(lines: &[&str]) -> Vec<String> {
    lines.iter().map(|line| line.to_string()).collect()
}

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let addr = listener.local_addr()?;
    let store: Store = Arc::default();
    let server = tokio::spawn(serve(listener, store));

    let tour = script(&[
        "SET name Ada Lovelace",
        "GET name",
        "INCR visits",
        "INCR visits",
        "SET word hello",
        "INCR word",
        "KEYS",
        "FLY away",
        "GET",
        "QUIT",
    ]);
    for (command, reply) in tour.iter().zip(session(addr, &tour).await?) {
        println!("{command:<22} -> {reply}");
    }

    let mut workers = Vec::new();
    for _ in 0..10 {
        let mut commands = vec!["INCR counter".to_string(); 20];
        commands.push("QUIT".to_string());
        workers.push(tokio::spawn(async move { session(addr, &commands).await }));
    }
    for worker in workers {
        worker.await.expect("worker panicked")?;
    }

    let check = script(&["GET counter", "COUNT", "KEYS", "QUIT"]);
    println!("after 10 clients x 20 INCR:");
    for (command, reply) in check.iter().zip(session(addr, &check).await?) {
        println!("  {command:<14} -> {reply}");
    }

    server.abort();
    Ok(())
}

Output

SET name Ada Lovelace  -> OK
GET name               -> VALUE Ada Lovelace
INCR visits            -> VALUE 1
INCR visits            -> VALUE 2
SET word hello         -> OK
INCR word              -> ERR value is not an integer
KEYS                   -> KEYS name visits word
FLY away               -> ERR bad command 'FLY away'
GET                    -> ERR bad command 'GET'
QUIT                   -> BYE
after 10 clients x 20 INCR:
  GET counter    -> VALUE 200
  COUNT          -> COUNT 4
  KEYS           -> KEYS counter name visits word
  QUIT           -> BYE

If the read and the write in INCR used two separate lock acquisitions, two clients could read the same number and both write the same result, and the counter would end below 200. Holding one guard across both steps prevents that. The guard is released before the reply is written, so no lock is held across an .await.

Part	Role
`Command` and `parse`	turn a line into a command or an error message
`execute`	applies a command to the map and returns the reply text
`Store`	`Arc<Mutex<HashMap>>` shared by every connection
`handle`	line loop: read a command, lock and execute, unlock, write the reply
`serve`	accept loop with one task per connection
`session`	client helper that sends a script in one write and reads replies until close

Extensions:

Add SETEX key seconds value and remove the key after the delay. Use a spawned task with sleep, and compare a version stamp so an older timer does not delete a newer value.
Reject lines longer than 1024 bytes. Replace lines() with a bounded read and reply ERR line too long.
Add MGET key1 key2 ... that returns all values in one reply line, taking the lock once.

Project 12: Chat room server

A chat room sends every message from one client to all the others. The server must read from the socket and deliver other people's messages at the same time, for every client.

Requirements:

A client connects, sends a nickname as its first line, and then sends chat lines.
Every chat line reaches all other connected clients as nickname: text. The sender does not receive its own line.
The server announces * nickname joined and * nickname left.
Nicknames are unique. A duplicate or invalid nickname gets an error and the connection closes.
/users lists current nicknames to the requester. /quit leaves the room.

Stage 1: fan-out of raw lines

Each connection subscribes to a broadcast channel. Its task waits on two things with select!: a line from its own client, and a message from the channel. Lines::next_line and broadcast::Receiver::recv are both cancel safe, so the branch that loses the race loses no data.

Each message carries the id of the connection that sent it, and every connection skips messages that carry its own id. The server writes READY after it has subscribed. A client waits for READY before it does anything else, which guarantees that no later message can be missed.

Rust

use std::net::SocketAddr;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, Lines};
use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::broadcast;

async fn handle(
    stream: TcpStream,
    id: usize,
    tx: broadcast::Sender<(usize, String)>,
) -> std::io::Result<()> {
    let mut rx = tx.subscribe();
    let (reader, mut writer) = stream.into_split();
    let mut lines = BufReader::new(reader).lines();
    writer.write_all(b"READY\n").await?;

    loop {
        tokio::select! {
            line = lines.next_line() => match line? {
                Some(text) => {
                    let _ = tx.send((id, text));
                }
                None => break,
            },
            received = rx.recv() => match received {
                Ok((from, text)) if from != id => {
                    writer.write_all(format!("{text}\n").as_bytes()).await?;
                }
                Ok(_) => {}
                Err(broadcast::error::RecvError::Lagged(_)) => {}
                Err(broadcast::error::RecvError::Closed) => break,
            },
        }
    }
    Ok(())
}

async fn serve(listener: TcpListener) {
    let (tx, _rx) = broadcast::channel(64);
    let mut next_id = 0;
    loop {
        let Ok((stream, _peer)) = listener.accept().await else {
            continue;
        };
        next_id += 1;
        let tx = tx.clone();
        tokio::spawn(async move {
            let _ = handle(stream, next_id, tx).await;
        });
    }
}

struct Client {
    lines: Lines<BufReader<OwnedReadHalf>>,
    writer: OwnedWriteHalf,
}

impl Client {
    async fn connect(addr: SocketAddr) -> std::io::Result<Client> {
        let (reader, writer) = TcpStream::connect(addr).await?.into_split();
        let mut client = Client { lines: BufReader::new(reader).lines(), writer };
        assert_eq!(client.next().await?.as_deref(), Some("READY"));
        Ok(client)
    }

    async fn say(&mut self, text: &str) -> std::io::Result<()> {
        self.writer.write_all(format!("{text}\n").as_bytes()).await
    }

    async fn next(&mut self) -> std::io::Result<Option<String>> {
        self.lines.next_line().await
    }
}

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let addr = listener.local_addr()?;
    let server = tokio::spawn(serve(listener));

    let mut one = Client::connect(addr).await?;
    let mut two = Client::connect(addr).await?;
    let mut three = Client::connect(addr).await?;

    one.say("hello everyone").await?;
    println!("two heard: {:?}", two.next().await?);
    println!("three heard: {:?}", three.next().await?);

    two.say("hi one").await?;
    println!("one heard: {:?}", one.next().await?);
    println!("three heard: {:?}", three.next().await?);

    server.abort();
    Ok(())
}

Output

two heard: Some("hello everyone")
three heard: Some("hello everyone")
one heard: Some("hi one")
three heard: Some("hi one")

Client one sent the first line, and the next line it received was the reply from two. Its own line never came back, because its handler skipped messages with its own id. tx.send returns an error when no receiver exists, and the handlers ignore it.

Stage 2: nicknames, joins, and leaves

Messages become an enum. A Message value knows how to render itself as a line and which nickname it belongs to. The handler asks for a nickname first: it writes NICK?, reads one line, subscribes, sends WELCOME nickname, and then broadcasts Joined. When the loop ends, it broadcasts Left. The client helper now takes a nickname.

Rust

use std::net::SocketAddr;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, Lines};
use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::broadcast;

#[derive(Clone, Debug)]
enum Message {
    Joined(String),
    Left(String),
    Chat { from: String, text: String },
}

impl Message {
    fn render(&self) -> String {
        match self {
            Message::Joined(nick) => format!("* {nick} joined"),
            Message::Left(nick) => format!("* {nick} left"),
            Message::Chat { from, text } => format!("{from}: {text}"),
        }
    }

    fn is_about(&self, me: &str) -> bool {
        match self {
            Message::Joined(nick) | Message::Left(nick) => nick == me,
            Message::Chat { from, .. } => from == me,
        }
    }
}

async fn chat(
    mut lines: Lines<BufReader<OwnedReadHalf>>,
    mut writer: OwnedWriteHalf,
    mut rx: broadcast::Receiver<Message>,
    tx: &broadcast::Sender<Message>,
    nick: &str,
) -> std::io::Result<()> {
    loop {
        tokio::select! {
            line = lines.next_line() => match line? {
                Some(text) if text.trim() == "/quit" => break,
                Some(text) => {
                    let _ = tx.send(Message::Chat { from: nick.to_string(), text });
                }
                None => break,
            },
            received = rx.recv() => match received {
                Ok(message) if !message.is_about(nick) => {
                    writer.write_all(format!("{}\n", message.render()).as_bytes()).await?;
                }
                Ok(_) => {}
                Err(broadcast::error::RecvError::Lagged(_)) => {}
                Err(broadcast::error::RecvError::Closed) => break,
            },
        }
    }
    Ok(())
}

async fn handle(stream: TcpStream, tx: broadcast::Sender<Message>) -> std::io::Result<()> {
    let (reader, mut writer) = stream.into_split();
    let mut lines = BufReader::new(reader).lines();
    writer.write_all(b"NICK?\n").await?;
    let Some(nick) = lines.next_line().await? else {
        return Ok(());
    };
    let nick = nick.trim().to_string();

    let rx = tx.subscribe();
    writer.write_all(format!("WELCOME {nick}\n").as_bytes()).await?;
    let _ = tx.send(Message::Joined(nick.clone()));

    let result = chat(lines, writer, rx, &tx, &nick).await;
    let _ = tx.send(Message::Left(nick));
    result
}

async fn serve(listener: TcpListener) {
    let (tx, _rx) = broadcast::channel(64);
    loop {
        let Ok((stream, _peer)) = listener.accept().await else {
            continue;
        };
        let tx = tx.clone();
        tokio::spawn(async move {
            let _ = handle(stream, tx).await;
        });
    }
}

struct Client {
    lines: Lines<BufReader<OwnedReadHalf>>,
    writer: OwnedWriteHalf,
}

impl Client {
    async fn connect(addr: SocketAddr, nick: &str) -> std::io::Result<Client> {
        let (reader, writer) = TcpStream::connect(addr).await?.into_split();
        let mut client = Client { lines: BufReader::new(reader).lines(), writer };
        assert_eq!(client.next().await?.as_deref(), Some("NICK?"));
        client.say(nick).await?;
        let welcome = client.next().await?;
        assert_eq!(welcome, Some(format!("WELCOME {nick}")));
        Ok(client)
    }

    async fn say(&mut self, text: &str) -> std::io::Result<()> {
        self.writer.write_all(format!("{text}\n").as_bytes()).await
    }

    async fn next(&mut self) -> std::io::Result<Option<String>> {
        self.lines.next_line().await
    }
}

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let addr = listener.local_addr()?;
    let server = tokio::spawn(serve(listener));

    let mut alice = Client::connect(addr, "alice").await?;
    let mut bob = Client::connect(addr, "bob").await?;
    println!("alice sees: {:?}", alice.next().await?);

    bob.say("hi alice").await?;
    println!("alice sees: {:?}", alice.next().await?);

    alice.say("hello bob").await?;
    println!("bob sees: {:?}", bob.next().await?);

    bob.say("/quit").await?;
    println!("alice sees: {:?}", alice.next().await?);
    println!("bob's connection closed: {}", bob.next().await?.is_none());

    server.abort();
    Ok(())
}

Output

alice sees: Some("* bob joined")
alice sees: Some("bob: hi alice")
bob sees: Some("alice: hello bob")
alice sees: Some("* bob left")
bob's connection closed: true

The WELCOME line is written after subscribe, so a client that has read it knows that every later message reaches it. Alice is already subscribed when Bob joins, so she receives * bob joined. Bob does not see his own join line, because is_about filters it. The handler broadcasts Left in every exit path, because it runs after chat returns and does not depend on how the loop ended.

Stage 3: unique nicknames, `/users`, and lag handling

The room state gains a set of taken nicknames in Arc<Mutex<HashSet<String>>>. A nickname is valid when it has 1 to 16 characters and no spaces. HashSet::insert returns false for a duplicate, so validation and reservation happen in one locked step. /users answers the requester alone by writing to its own socket. A receiver that falls behind gets RecvError::Lagged(n) from the channel, and the handler tells the client how many messages were skipped.

Rust

use std::collections::HashSet;
use std::net::SocketAddr;
use std::sync::{Arc, Mutex};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, Lines};
use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::broadcast;

#[derive(Clone, Debug)]
enum Message {
    Joined(String),
    Left(String),
    Chat { from: String, text: String },
}

impl Message {
    fn render(&self) -> String {
        match self {
            Message::Joined(nick) => format!("* {nick} joined"),
            Message::Left(nick) => format!("* {nick} left"),
            Message::Chat { from, text } => format!("{from}: {text}"),
        }
    }

    fn is_about(&self, me: &str) -> bool {
        match self {
            Message::Joined(nick) | Message::Left(nick) => nick == me,
            Message::Chat { from, .. } => from == me,
        }
    }
}

#[derive(Clone)]
struct Room {
    tx: broadcast::Sender<Message>,
    users: Arc<Mutex<HashSet<String>>>,
}

fn valid_nick(nick: &str) -> bool {
    !nick.is_empty() && nick.chars().count() <= 16 && !nick.contains(char::is_whitespace)
}

async fn chat(
    mut lines: Lines<BufReader<OwnedReadHalf>>,
    mut writer: OwnedWriteHalf,
    mut rx: broadcast::Receiver<Message>,
    room: &Room,
    nick: &str,
) -> std::io::Result<()> {
    loop {
        tokio::select! {
            line = lines.next_line() => {
                let Some(text) = line? else { break };
                let text = text.trim().to_string();
                if text == "/quit" {
                    break;
                } else if text == "/users" {
                    let mut names: Vec<String> =
                        room.users.lock().unwrap().iter().cloned().collect();
                    names.sort();
                    writer.write_all(format!("* users: {}\n", names.join(", ")).as_bytes()).await?;
                } else if !text.is_empty() {
                    let _ = room.tx.send(Message::Chat { from: nick.to_string(), text });
                }
            }
            received = rx.recv() => match received {
                Ok(message) if !message.is_about(nick) => {
                    writer.write_all(format!("{}\n", message.render()).as_bytes()).await?;
                }
                Ok(_) => {}
                Err(broadcast::error::RecvError::Lagged(missed)) => {
                    writer.write_all(format!("* you missed {missed} messages\n").as_bytes()).await?;
                }
                Err(broadcast::error::RecvError::Closed) => break,
            },
        }
    }
    Ok(())
}

async fn handle(stream: TcpStream, room: Room) -> std::io::Result<()> {
    let (reader, mut writer) = stream.into_split();
    let mut lines = BufReader::new(reader).lines();
    writer.write_all(b"NICK?\n").await?;
    let Some(nick) = lines.next_line().await? else {
        return Ok(());
    };
    let nick = nick.trim().to_string();

    let reserved = valid_nick(&nick) && room.users.lock().unwrap().insert(nick.clone());
    if !reserved {
        writer.write_all(b"ERR nickname taken or invalid\n").await?;
        return Ok(());
    }

    let rx = room.tx.subscribe();
    let result = match writer.write_all(format!("WELCOME {nick}\n").as_bytes()).await {
        Ok(()) => {
            let _ = room.tx.send(Message::Joined(nick.clone()));
            chat(lines, writer, rx, &room, &nick).await
        }
        Err(e) => Err(e),
    };

    room.users.lock().unwrap().remove(&nick);
    let _ = room.tx.send(Message::Left(nick));
    result
}

async fn serve(listener: TcpListener) {
    let (tx, _rx) = broadcast::channel(64);
    let room = Room { tx, users: Arc::default() };
    loop {
        let Ok((stream, _peer)) = listener.accept().await else {
            continue;
        };
        let room = room.clone();
        tokio::spawn(async move {
            let _ = handle(stream, room).await;
        });
    }
}

struct Client {
    lines: Lines<BufReader<OwnedReadHalf>>,
    writer: OwnedWriteHalf,
}

impl Client {
    async fn open(addr: SocketAddr, nick: &str) -> std::io::Result<Client> {
        let (reader, writer) = TcpStream::connect(addr).await?.into_split();
        let mut client = Client { lines: BufReader::new(reader).lines(), writer };
        assert_eq!(client.next().await?.as_deref(), Some("NICK?"));
        client.say(nick).await?;
        Ok(client)
    }

    async fn connect(addr: SocketAddr, nick: &str) -> std::io::Result<Client> {
        let mut client = Client::open(addr, nick).await?;
        assert_eq!(client.next().await?, Some(format!("WELCOME {nick}")));
        Ok(client)
    }

    async fn say(&mut self, text: &str) -> std::io::Result<()> {
        self.writer.write_all(format!("{text}\n").as_bytes()).await
    }

    async fn next(&mut self) -> std::io::Result<Option<String>> {
        self.lines.next_line().await
    }
}

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let addr = listener.local_addr()?;
    let server = tokio::spawn(serve(listener));

    let mut alice = Client::connect(addr, "alice").await?;
    let mut bob = Client::connect(addr, "bob").await?;
    println!("alice sees: {:?}", alice.next().await?);

    let mut duplicate = Client::open(addr, "alice").await?;
    println!("duplicate gets: {:?}", duplicate.next().await?);
    println!("duplicate closed: {}", duplicate.next().await?.is_none());

    let mut spaced = Client::open(addr, "two words").await?;
    println!("invalid gets: {:?}", spaced.next().await?);

    bob.say("hi alice").await?;
    println!("alice sees: {:?}", alice.next().await?);

    alice.say("hello bob").await?;
    println!("bob sees: {:?}", bob.next().await?);

    alice.say("/users").await?;
    println!("alice sees: {:?}", alice.next().await?);

    bob.say("/quit").await?;
    println!("alice sees: {:?}", alice.next().await?);
    println!("bob's connection closed: {}", bob.next().await?.is_none());

    alice.say("/users").await?;
    println!("alice sees: {:?}", alice.next().await?);

    server.abort();
    Ok(())
}

Output

alice sees: Some("* bob joined")
duplicate gets: Some("ERR nickname taken or invalid")
duplicate closed: true
invalid gets: Some("ERR nickname taken or invalid")
alice sees: Some("bob: hi alice")
bob sees: Some("alice: hello bob")
alice sees: Some("* users: alice, bob")
alice sees: Some("* bob left")
bob's connection closed: true
alice sees: Some("* users: alice")

The failed alice connection is closed and announces nothing, so the room does not see it. After Bob leaves, his nickname is removed from the set, which the final /users shows. The Left message is sent after the removal, so a client that receives * bob left can rely on the nickname being free again.

Part	Role
`Message`	the events in the room: chat lines, joins, and leaves, each with a text form
`Room`	the broadcast sender and the shared set of nicknames
`handle`	asks for a nickname, reserves it, runs the chat loop, always cleans up
`chat`	`select!` over the client's lines and the room's messages
`valid_nick`	length and whitespace check before the nickname is reserved
`Client`	test client that reads and writes lines

Extensions:

Add /msg nickname text for a private message. Keep a HashMap<String, mpsc::Sender<String>> of per-user queues next to the broadcast channel.
Limit each client to five chat lines per second. Count lines in the current second, drop the extra lines, and reply * slow down.
Keep the last 20 chat lines in a VecDeque inside the room and send them to each new client after WELCOME.

Project 13: Minimal HTTP/1.1 server

HTTP/1.1 is a text protocol over TCP. A request is a request line, header lines, an empty line, and an optional body whose length is given by Content-Length. A response is a status line, headers, an empty line, and a body. Building the parser and the writer by hand shows where the protocol's framing rules sit.

Requirements:

Parse the request line, the headers (names are case-insensitive), and a body of Content-Length bytes.
Routes: GET /, GET /health, GET /hello/{name}, and POST /echo.
Reply 404 for unknown paths, 405 for a known path with the wrong method, 400 for malformed requests, and 413 for a body larger than 1024 bytes.
Every response carries Content-Length, so the client knows where the body ends.
Keep the connection open for more requests unless the client sends Connection: close.
Handler code works on any stream that implements AsyncRead and AsyncWrite, so tests can use tokio::io::duplex instead of the network.

Stage 1: parse a request and write a response

read_request is generic over the AsyncBufRead trait, which BufReader implements. It returns Ok(None) when the stream ends before a request line, which is how a client closes a kept-open connection. The handler is generic over the stream type. The test connects a duplex pipe, which behaves like a socket but lives in memory.

Rust

use std::collections::HashMap;
use tokio::io::{
    AsyncBufRead, AsyncBufReadExt, AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, BufReader,
};

const MAX_BODY: usize = 1024;

#[derive(Debug)]
struct Request {
    method: String,
    path: String,
    headers: HashMap<String, String>,
    body: Vec<u8>,
}

#[derive(Debug)]
enum HttpError {
    BadRequest(&'static str),
    BodyTooLarge,
    Io(std::io::Error),
}

impl From<std::io::Error> for HttpError {
    fn from(e: std::io::Error) -> Self {
        HttpError::Io(e)
    }
}

async fn read_request<R: AsyncBufRead + Unpin>(
    reader: &mut R,
) -> Result<Option<Request>, HttpError> {
    let mut line = String::new();
    if reader.read_line(&mut line).await? == 0 {
        return Ok(None);
    }
    let mut parts = line.split_whitespace();
    let (Some(method), Some(path), Some(version), None) =
        (parts.next(), parts.next(), parts.next(), parts.next())
    else {
        return Err(HttpError::BadRequest("malformed request line"));
    };
    if version != "HTTP/1.1" {
        return Err(HttpError::BadRequest("unsupported version"));
    }

    let mut headers = HashMap::new();
    loop {
        let mut header = String::new();
        if reader.read_line(&mut header).await? == 0 {
            return Err(HttpError::BadRequest("connection closed inside headers"));
        }
        let header = header.trim_end();
        if header.is_empty() {
            break;
        }
        if headers.len() >= 32 {
            return Err(HttpError::BadRequest("too many headers"));
        }
        let Some((name, value)) = header.split_once(':') else {
            return Err(HttpError::BadRequest("malformed header"));
        };
        headers.insert(name.trim().to_ascii_lowercase(), value.trim().to_string());
    }

    let length = match headers.get("content-length") {
        Some(value) => value
            .parse::<usize>()
            .map_err(|_| HttpError::BadRequest("bad content-length"))?,
        None => 0,
    };
    if length > MAX_BODY {
        return Err(HttpError::BodyTooLarge);
    }
    let mut body = vec![0u8; length];
    reader.read_exact(&mut body).await?;

    Ok(Some(Request {
        method: method.to_string(),
        path: path.to_string(),
        headers,
        body,
    }))
}

async fn serve_one<S: AsyncRead + AsyncWrite + Unpin>(stream: S) -> std::io::Result<()> {
    let (reader, mut writer) = tokio::io::split(stream);
    let mut reader = BufReader::new(reader);

    let body = match read_request(&mut reader).await {
        Ok(Some(request)) => {
            let mut names: Vec<_> = request.headers.keys().cloned().collect();
            names.sort();
            format!(
                "method={} path={} headers={:?} body={} bytes\n",
                request.method,
                request.path,
                names,
                request.body.len()
            )
        }
        Ok(None) => return Ok(()),
        Err(HttpError::Io(e)) => return Err(e),
        Err(other) => format!("error: {other:?}\n"),
    };

    let response = format!(
        "HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
        body.len(),
        body
    );
    writer.write_all(response.as_bytes()).await?;
    writer.shutdown().await
}

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let (mut client, server) = tokio::io::duplex(16 * 1024);
    let handle = tokio::spawn(serve_one(server));

    let request = "POST /echo?x=1 HTTP/1.1\r\nHost: example.test\r\nContent-Length: 5\r\nX-Custom: Yes\r\n\r\nhello";
    client.write_all(request.as_bytes()).await?;

    let mut raw = String::new();
    client.read_to_string(&mut raw).await?;
    handle.await.expect("server panicked")?;

    println!("{raw:?}");
    Ok(())
}

Output

"HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nContent-Length: 87\r\nConnection: close\r\n\r\nmethod=POST path=/echo?x=1 headers=[\"content-length\", \"host\", \"x-custom\"] body=5 bytes\n"

The output shows the response as a debug string so that the \r\n line endings are visible. HTTP lines end with carriage return and line feed, and the empty line between headers and body is \r\n\r\n. The parser lowercases header names, so Content-Length and content-length are the same header. The query string ?x=1 stays part of path; the router in the next stage removes it.

Stage 2: routing and error responses

Routing is a plain function from Request to Response, so it is testable without I/O. Parse failures map to 400 and 413 responses that also close the connection, because after a framing error the server cannot know where the next request starts. The test helper roundtrip sends raw bytes through a fresh duplex pipe and returns the status line and body.

Rust

use std::collections::HashMap;
use tokio::io::{
    AsyncBufRead, AsyncBufReadExt, AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, BufReader,
};

const MAX_BODY: usize = 1024;

#[derive(Debug)]
struct Request {
    method: String,
    path: String,
    headers: HashMap<String, String>,
    body: Vec<u8>,
}

#[derive(Debug)]
enum HttpError {
    BadRequest(&'static str),
    BodyTooLarge,
    Io(std::io::Error),
}

impl From<std::io::Error> for HttpError {
    fn from(e: std::io::Error) -> Self {
        HttpError::Io(e)
    }
}

struct Response {
    status: u16,
    reason: &'static str,
    body: Vec<u8>,
}

impl Response {
    fn text(status: u16, reason: &'static str, body: impl Into<String>) -> Response {
        Response { status, reason, body: body.into().into_bytes() }
    }

    fn to_bytes(&self) -> Vec<u8> {
        let head = format!(
            "HTTP/1.1 {} {}\r\nContent-Type: text/plain\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
            self.status,
            self.reason,
            self.body.len()
        );
        let mut bytes = head.into_bytes();
        bytes.extend_from_slice(&self.body);
        bytes
    }
}

async fn read_request<R: AsyncBufRead + Unpin>(
    reader: &mut R,
) -> Result<Option<Request>, HttpError> {
    let mut line = String::new();
    if reader.read_line(&mut line).await? == 0 {
        return Ok(None);
    }
    let mut parts = line.split_whitespace();
    let (Some(method), Some(path), Some(version), None) =
        (parts.next(), parts.next(), parts.next(), parts.next())
    else {
        return Err(HttpError::BadRequest("malformed request line"));
    };
    if version != "HTTP/1.1" {
        return Err(HttpError::BadRequest("unsupported version"));
    }

    let mut headers = HashMap::new();
    loop {
        let mut header = String::new();
        if reader.read_line(&mut header).await? == 0 {
            return Err(HttpError::BadRequest("connection closed inside headers"));
        }
        let header = header.trim_end();
        if header.is_empty() {
            break;
        }
        if headers.len() >= 32 {
            return Err(HttpError::BadRequest("too many headers"));
        }
        let Some((name, value)) = header.split_once(':') else {
            return Err(HttpError::BadRequest("malformed header"));
        };
        headers.insert(name.trim().to_ascii_lowercase(), value.trim().to_string());
    }

    let length = match headers.get("content-length") {
        Some(value) => value
            .parse::<usize>()
            .map_err(|_| HttpError::BadRequest("bad content-length"))?,
        None => 0,
    };
    if length > MAX_BODY {
        return Err(HttpError::BodyTooLarge);
    }
    let mut body = vec![0u8; length];
    reader.read_exact(&mut body).await?;

    Ok(Some(Request {
        method: method.to_string(),
        path: path.to_string(),
        headers,
        body,
    }))
}

fn route(request: &Request) -> Response {
    let path = request.path.split('?').next().unwrap_or("");
    let get = request.method == "GET";

    if let Some(name) = path.strip_prefix("/hello/") {
        if !name.is_empty() {
            return if get {
                Response::text(200, "OK", format!("Hello, {name}!\n"))
            } else {
                Response::text(405, "Method Not Allowed", "use GET\n")
            };
        }
    }

    match path {
        "/" if get => Response::text(200, "OK", "Welcome to the minimal server\n"),
        "/health" if get => Response::text(200, "OK", "ok\n"),
        "/echo" if request.method == "POST" => Response {
            status: 200,
            reason: "OK",
            body: request.body.clone(),
        },
        "/" | "/health" => Response::text(405, "Method Not Allowed", "use GET\n"),
        "/echo" => Response::text(405, "Method Not Allowed", "use POST\n"),
        _ => Response::text(404, "Not Found", "no such route\n"),
    }
}

async fn serve_one<S: AsyncRead + AsyncWrite + Unpin>(stream: S) -> std::io::Result<()> {
    let (reader, mut writer) = tokio::io::split(stream);
    let mut reader = BufReader::new(reader);

    let response = match read_request(&mut reader).await {
        Ok(Some(request)) => route(&request),
        Ok(None) => return Ok(()),
        Err(HttpError::BadRequest(why)) => {
            Response::text(400, "Bad Request", format!("{why}\n"))
        }
        Err(HttpError::BodyTooLarge) => {
            Response::text(413, "Payload Too Large", format!("limit is {MAX_BODY} bytes\n"))
        }
        Err(HttpError::Io(e)) => return Err(e),
    };
    writer.write_all(&response.to_bytes()).await?;
    writer.shutdown().await
}

async fn roundtrip(raw: &str) -> String {
    let (mut client, server) = tokio::io::duplex(16 * 1024);
    let handle = tokio::spawn(serve_one(server));
    client.write_all(raw.as_bytes()).await.expect("write request");

    let mut response = String::new();
    client.read_to_string(&mut response).await.expect("read response");
    handle.await.expect("server panicked").expect("server error");

    let (head, body) = response.split_once("\r\n\r\n").expect("blank line");
    let status = head.lines().next().expect("status line");
    format!("{status} | {}", body.trim_end())
}

#[tokio::main]
async fn main() {
    let cases = [
        ("GET /", "GET / HTTP/1.1\r\n\r\n"),
        ("GET /health", "GET /health HTTP/1.1\r\nHost: test\r\n\r\n"),
        ("GET /hello/Ferris?lang=en", "GET /hello/Ferris?lang=en HTTP/1.1\r\n\r\n"),
        ("POST /echo", "POST /echo HTTP/1.1\r\nContent-Length: 11\r\n\r\nround trip!"),
        ("GET /nowhere", "GET /nowhere HTTP/1.1\r\n\r\n"),
        ("POST /", "POST / HTTP/1.1\r\nContent-Length: 0\r\n\r\n"),
        ("GET /echo", "GET /echo HTTP/1.1\r\n\r\n"),
        ("no version", "NONSENSE\r\n\r\n"),
        ("HTTP/1.0", "GET / HTTP/1.0\r\n\r\n"),
        ("header without colon", "GET / HTTP/1.1\r\nno colon here\r\n\r\n"),
        ("length abc", "POST /echo HTTP/1.1\r\nContent-Length: abc\r\n\r\n"),
        ("length 5000", "POST /echo HTTP/1.1\r\nContent-Length: 5000\r\n\r\n"),
    ];
    for (label, raw) in cases {
        println!("{label:<26} => {}", roundtrip(raw).await);
    }
}

Output

GET / HTTP/1.1                   => HTTP/1.1 200 OK | Welcome to the minimal server
GET /health HTTP/1.1             => HTTP/1.1 200 OK | ok
GET /hello/Ferris?lang=en HTTP/1.1 => HTTP/1.1 200 OK | Hello, Ferris!
POST /echo HTTP/1.1              => HTTP/1.1 200 OK | round trip!
GET /nowhere HTTP/1.1            => HTTP/1.1 404 Not Found | no such route
POST / HTTP/1.1                  => HTTP/1.1 405 Method Not Allowed | use GET
GET /echo HTTP/1.1               => HTTP/1.1 405 Method Not Allowed | use POST
NONSENSE                         => HTTP/1.1 400 Bad Request | malformed request line
GET / HTTP/1.0                   => HTTP/1.1 400 Bad Request | unsupported version
GET / HTTP/1.1                   => HTTP/1.1 400 Bad Request | malformed header
POST /echo HTTP/1.1              => HTTP/1.1 400 Bad Request | bad content-length
POST /echo HTTP/1.1              => HTTP/1.1 413 Payload Too Large | limit is 1024 bytes

Each case runs on its own in-memory connection, so the test needs no port and no timing. The 413 case sends only the header. The server rejects it from the Content-Length value without reading a body, and refuses to allocate a buffer of 5000 bytes for a limit of 1024. Without that check, a client could make the server allocate any amount of memory by claiming a large length.

Stage 3: TCP, keep-alive, and shared state

The final program serves real connections. serve loops over requests on one connection until the client sends Connection: close, closes its side, or stays idle for five seconds. Responses now say Connection: keep-alive or Connection: close. A request counter behind Arc<Mutex<_>> backs the GET /stats route. The client reads responses by the rule the server follows: the status line, headers up to the empty line, then exactly Content-Length bytes.

The stream generics from stage 2 stay, so serve still works with duplex in tests, and main runs it on TCP.

Rust

use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio::io::{
    AsyncBufRead, AsyncBufReadExt, AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, BufReader,
};
use tokio::net::{TcpListener, TcpStream};
use tokio::time::timeout;

const MAX_BODY: usize = 1024;
const IDLE_TIMEOUT: Duration = Duration::from_secs(5);

#[derive(Debug)]
struct Request {
    method: String,
    path: String,
    headers: HashMap<String, String>,
    body: Vec<u8>,
}

#[derive(Debug)]
enum HttpError {
    BadRequest(&'static str),
    BodyTooLarge,
    Io(std::io::Error),
}

impl From<std::io::Error> for HttpError {
    fn from(e: std::io::Error) -> Self {
        HttpError::Io(e)
    }
}

struct Response {
    status: u16,
    reason: &'static str,
    body: Vec<u8>,
    close: bool,
}

impl Response {
    fn text(status: u16, reason: &'static str, body: impl Into<String>) -> Response {
        Response { status, reason, body: body.into().into_bytes(), close: false }
    }

    fn closing(mut self) -> Response {
        self.close = true;
        self
    }

    fn to_bytes(&self) -> Vec<u8> {
        let connection = if self.close { "close" } else { "keep-alive" };
        let head = format!(
            "HTTP/1.1 {} {}\r\nContent-Type: text/plain\r\nContent-Length: {}\r\nConnection: {}\r\n\r\n",
            self.status,
            self.reason,
            self.body.len(),
            connection
        );
        let mut bytes = head.into_bytes();
        bytes.extend_from_slice(&self.body);
        bytes
    }
}

#[derive(Default)]
struct State {
    requests: Mutex<u64>,
}

async fn read_request<R: AsyncBufRead + Unpin>(
    reader: &mut R,
) -> Result<Option<Request>, HttpError> {
    let mut line = String::new();
    if reader.read_line(&mut line).await? == 0 {
        return Ok(None);
    }
    let mut parts = line.split_whitespace();
    let (Some(method), Some(path), Some(version), None) =
        (parts.next(), parts.next(), parts.next(), parts.next())
    else {
        return Err(HttpError::BadRequest("malformed request line"));
    };
    if version != "HTTP/1.1" {
        return Err(HttpError::BadRequest("unsupported version"));
    }

    let mut headers = HashMap::new();
    loop {
        let mut header = String::new();
        if reader.read_line(&mut header).await? == 0 {
            return Err(HttpError::BadRequest("connection closed inside headers"));
        }
        let header = header.trim_end();
        if header.is_empty() {
            break;
        }
        if headers.len() >= 32 {
            return Err(HttpError::BadRequest("too many headers"));
        }
        let Some((name, value)) = header.split_once(':') else {
            return Err(HttpError::BadRequest("malformed header"));
        };
        headers.insert(name.trim().to_ascii_lowercase(), value.trim().to_string());
    }

    let length = match headers.get("content-length") {
        Some(value) => value
            .parse::<usize>()
            .map_err(|_| HttpError::BadRequest("bad content-length"))?,
        None => 0,
    };
    if length > MAX_BODY {
        return Err(HttpError::BodyTooLarge);
    }
    let mut body = vec![0u8; length];
    reader.read_exact(&mut body).await?;

    Ok(Some(Request {
        method: method.to_string(),
        path: path.to_string(),
        headers,
        body,
    }))
}

fn route(request: &Request, state: &State) -> Response {
    let path = request.path.split('?').next().unwrap_or("");
    let get = request.method == "GET";

    if let Some(name) = path.strip_prefix("/hello/") {
        if !name.is_empty() {
            return if get {
                Response::text(200, "OK", format!("Hello, {name}!\n"))
            } else {
                Response::text(405, "Method Not Allowed", "use GET\n")
            };
        }
    }

    match path {
        "/" if get => Response::text(200, "OK", "Welcome to the minimal server\n"),
        "/health" if get => Response::text(200, "OK", "ok\n"),
        "/stats" if get => {
            let count = *state.requests.lock().unwrap();
            Response::text(200, "OK", format!("requests so far: {count}\n"))
        }
        "/echo" if request.method == "POST" => Response {
            status: 200,
            reason: "OK",
            body: request.body.clone(),
            close: false,
        },
        "/" | "/health" | "/stats" => Response::text(405, "Method Not Allowed", "use GET\n"),
        "/echo" => Response::text(405, "Method Not Allowed", "use POST\n"),
        _ => Response::text(404, "Not Found", "no such route\n"),
    }
}

async fn serve<S: AsyncRead + AsyncWrite + Unpin>(
    stream: S,
    state: Arc<State>,
) -> std::io::Result<()> {
    let (reader, mut writer) = tokio::io::split(stream);
    let mut reader = BufReader::new(reader);

    loop {
        let parsed = match timeout(IDLE_TIMEOUT, read_request(&mut reader)).await {
            Ok(parsed) => parsed,
            Err(_idle) => return Ok(()),
        };
        let response = match parsed {
            Ok(Some(request)) => {
                *state.requests.lock().unwrap() += 1;
                let wants_close = request
                    .headers
                    .get("connection")
                    .is_some_and(|value| value.eq_ignore_ascii_case("close"));
                let response = route(&request, &state);
                if wants_close { response.closing() } else { response }
            }
            Ok(None) => return Ok(()),
            Err(HttpError::BadRequest(why)) => {
                Response::text(400, "Bad Request", format!("{why}\n")).closing()
            }
            Err(HttpError::BodyTooLarge) => {
                Response::text(413, "Payload Too Large", format!("limit is {MAX_BODY} bytes\n"))
                    .closing()
            }
            Err(HttpError::Io(e)) => return Err(e),
        };
        writer.write_all(&response.to_bytes()).await?;
        if response.close {
            return writer.shutdown().await;
        }
    }
}

async fn accept_loop(listener: TcpListener, state: Arc<State>) {
    loop {
        let Ok((stream, _peer)) = listener.accept().await else {
            continue;
        };
        let state = Arc::clone(&state);
        tokio::spawn(async move {
            let _ = serve(stream, state).await;
        });
    }
}

async fn read_response<R: AsyncBufRead + Unpin>(
    reader: &mut R,
) -> std::io::Result<(String, String)> {
    let mut status = String::new();
    reader.read_line(&mut status).await?;
    let mut length = 0;
    loop {
        let mut header = String::new();
        reader.read_line(&mut header).await?;
        let header = header.trim_end();
        if header.is_empty() {
            break;
        }
        if let Some((name, value)) = header.split_once(':') {
            if name.eq_ignore_ascii_case("content-length") {
                length = value.trim().parse().expect("numeric length");
            }
        }
    }
    let mut body = vec![0u8; length];
    reader.read_exact(&mut body).await?;
    Ok((
        status.trim_end().to_string(),
        String::from_utf8_lossy(&body).trim_end().to_string(),
    ))
}

async fn one_shot(addr: std::net::SocketAddr, raw: &str) -> std::io::Result<String> {
    let mut stream = TcpStream::connect(addr).await?;
    stream.write_all(raw.as_bytes()).await?;
    let mut reader = BufReader::new(stream);
    let (status, body) = read_response(&mut reader).await?;
    Ok(format!("{status} | {body}"))
}

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let addr = listener.local_addr()?;
    let state = Arc::new(State::default());
    let server = tokio::spawn(accept_loop(listener, Arc::clone(&state)));

    println!("-- three requests on one connection");
    let stream = TcpStream::connect(addr).await?;
    let (read_half, mut write_half) = stream.into_split();
    let mut reader = BufReader::new(read_half);
    let requests = [
        "GET /hello/one HTTP/1.1\r\nHost: t\r\n\r\n",
        "POST /echo HTTP/1.1\r\nContent-Length: 4\r\n\r\nping",
        "GET /health HTTP/1.1\r\nConnection: close\r\n\r\n",
    ];
    for raw in requests {
        write_half.write_all(raw.as_bytes()).await?;
        let (status, body) = read_response(&mut reader).await?;
        println!("{status} | {body}");
    }
    let mut rest = String::new();
    let extra = reader.read_line(&mut rest).await?;
    println!("server closed the connection: {}", extra == 0);

    println!("-- three clients at once");
    let mut clients = Vec::new();
    for name in ["ann", "ben", "cy"] {
        clients.push(tokio::spawn(async move {
            let raw = format!("GET /hello/{name} HTTP/1.1\r\nConnection: close\r\n\r\n");
            one_shot(addr, &raw).await
        }));
    }
    for client in clients {
        println!("{}", client.await.expect("client panicked")?);
    }

    println!("-- errors");
    for raw in [
        "GET /missing HTTP/1.1\r\nConnection: close\r\n\r\n",
        "DELETE /health HTTP/1.1\r\nConnection: close\r\n\r\n",
        "GET / HTTP/2\r\n\r\n",
        "POST /echo HTTP/1.1\r\nContent-Length: 5000\r\n\r\n",
    ] {
        println!("{}", one_shot(addr, raw).await?);
    }

    println!("-- stats");
    println!(
        "{}",
        one_shot(addr, "GET /stats HTTP/1.1\r\nConnection: close\r\n\r\n").await?
    );

    server.abort();
    Ok(())
}

Output

-- three requests on one connection
HTTP/1.1 200 OK | Hello, one!
HTTP/1.1 200 OK | ping
HTTP/1.1 200 OK | ok
server closed the connection: true
-- three clients at once
HTTP/1.1 200 OK | Hello, ann!
HTTP/1.1 200 OK | Hello, ben!
HTTP/1.1 200 OK | Hello, cy!
-- errors
HTTP/1.1 404 Not Found | no such route
HTTP/1.1 405 Method Not Allowed | use GET
HTTP/1.1 400 Bad Request | unsupported version
HTTP/1.1 413 Payload Too Large | limit is 1024 bytes
-- stats
HTTP/1.1 200 OK | requests so far: 9

The three requests on the first connection are sent one at a time, and each response is read before the next request goes out. The server answers all three on the same connection because the first two carry no Connection: close. The third request asks for close, the server's response says Connection: close, and the server then shuts down its side, so the client's next read returns 0.

The request count is 9: three requests on the first connection, three concurrent clients, /missing, DELETE /health, and /stats itself. The counter includes only requests that parsed. The HTTP/2 request and the oversized POST fail inside read_request, return 400 and 413, and do not count.

Part	Role
`read_request`	request line, headers, and body by `Content-Length`, with limits
`HttpError`	separates client mistakes (400, 413) from I/O failures
`Response`	status, body, and connection policy, and the wire format
`route`	pure function from request to response, testable without I/O
`serve`	per-connection loop with keep-alive and an idle timeout, generic over the stream type
`accept_loop`	one task per TCP connection
`read_response`	client-side parser that uses `Content-Length` to find the end of the body

Extensions:

Add a limit for header lines. Read with take(8192) on the BufReader for each line and answer 431 when a line is longer.
Support Transfer-Encoding: chunked request bodies: read a hexadecimal size line, then that many bytes, until a size of 0.
Serve files from a directory with tokio::fs::read. Reject paths that contain .., and derive Content-Type from the extension.

Project 14: Process runner and file follower

A build tool or job runner starts programs, limits how long they run, captures their output, and shows a log while a job is still running. This project builds those pieces from tokio::process and tokio::fs.

Requirements:

Run a program with arguments and a time limit. Return its exit status, captured stdout, and captured stderr.
A program that exceeds the limit is killed and reported as timed out.
A program that cannot start is an error with a kind, not a panic.
A file follower prints new lines as they are appended to a file, including when the file does not exist yet and when a line arrives in two parts.
A job runner starts a program that writes to a log file, follows the log while the program runs, and reports the outcome.
Several jobs run at the same time.

Stage 1: run a command with a time limit

run_with_timeout starts a program with piped output and kill_on_drop(true). One future reads both streams at the same time and then waits for the exit status. timeout wraps that future. When the limit passes, the future is dropped, and the code kills the child and waits for it, so no process is left behind. Output that arrived before the timeout is lost in this version, because the buffers live inside the dropped future.

Rust

use std::process::Stdio;
use std::time::Duration;
use tokio::io::AsyncReadExt;
use tokio::process::Command;
use tokio::time::timeout;

#[derive(Debug)]
enum Outcome {
    Exited(i32),
    Signaled,
    TimedOut,
}

struct Report {
    outcome: Outcome,
    stdout: String,
    stderr: String,
}

async fn run_with_timeout(
    program: &str,
    args: &[&str],
    limit: Duration,
) -> std::io::Result<Report> {
    let mut child = Command::new(program)
        .args(args)
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .kill_on_drop(true)
        .spawn()?;
    let mut stdout = child.stdout.take().expect("stdout is piped");
    let mut stderr = child.stderr.take().expect("stderr is piped");

    let work = async {
        let mut out = String::new();
        let mut err = String::new();
        let (a, b) = tokio::join!(
            stdout.read_to_string(&mut out),
            stderr.read_to_string(&mut err)
        );
        a?;
        b?;
        let status = child.wait().await?;
        Ok::<_, std::io::Error>((out, err, status))
    };

    match timeout(limit, work).await {
        Ok(result) => {
            let (stdout, stderr, status) = result?;
            let outcome = match status.code() {
                Some(code) => Outcome::Exited(code),
                None => Outcome::Signaled,
            };
            Ok(Report { outcome, stdout, stderr })
        }
        Err(_elapsed) => {
            child.kill().await?;
            Ok(Report {
                outcome: Outcome::TimedOut,
                stdout: String::new(),
                stderr: String::new(),
            })
        }
    }
}

#[tokio::main]
async fn main() {
    let jobs: [(&str, &[&str], u64); 4] = [
        ("echo", &["hello", "runner"], 5000),
        ("sh", &["-c", "echo working; echo oops >&2; exit 2"], 5000),
        ("sleep", &["30"], 200),
        ("no-such-program-xyz", &[], 5000),
    ];

    for (program, args, limit_ms) in jobs {
        let limit = Duration::from_millis(limit_ms);
        match run_with_timeout(program, args, limit).await {
            Ok(report) => println!(
                "{program}: {:?}, stdout={:?}, stderr={:?}",
                report.outcome, report.stdout, report.stderr
            ),
            Err(e) => println!("{program}: could not run ({:?})", e.kind()),
        }
    }
}

Output

echo: Exited(0), stdout="hello runner\n", stderr=""
sh: Exited(2), stdout="working\n", stderr="oops\n"
sleep: TimedOut, stdout="", stderr=""
no-such-program-xyz: could not run (NotFound)

The time limits are real: sleep 30 is killed after 200 ms. The limits on the other jobs are 5 seconds, far above their run time, so the outcome does not depend on how busy the machine is. The failed spawn for the missing program returns an io::Error with kind NotFound, so run_with_timeout returns Err and the caller decides what to print.

Stage 2: follow a file

Follower remembers how many bytes it has already consumed and holds the bytes of an unfinished line. poll reads the whole file with fs::read, takes the bytes after the offset, and returns each complete line. A file that does not exist yields no lines. A file shorter than the offset was truncated, so the follower starts from the beginning. follow calls poll on a 10 ms interval until a oneshot message says the writer is done, and then polls one last time to pick up the final bytes.

The demo writer appends alpha, then be, then ta, then gamma, with pauses, and creates the file only after the follower has started.

Rust

use std::path::{Path, PathBuf};
use std::time::Duration;
use tokio::fs::{self, OpenOptions};
use tokio::io::AsyncWriteExt;
use tokio::sync::oneshot;
use tokio::time::{interval, sleep};

struct Follower {
    path: PathBuf,
    offset: usize,
    partial: Vec<u8>,
}

impl Follower {
    fn new(path: PathBuf) -> Follower {
        Follower { path, offset: 0, partial: Vec::new() }
    }

    async fn poll(&mut self) -> std::io::Result<Vec<String>> {
        let bytes = match fs::read(&self.path).await {
            Ok(bytes) => bytes,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
            Err(e) => return Err(e),
        };
        if bytes.len() < self.offset {
            self.offset = 0;
            self.partial.clear();
        }
        self.partial.extend_from_slice(&bytes[self.offset..]);
        self.offset = bytes.len();

        let mut lines = Vec::new();
        while let Some(end) = self.partial.iter().position(|&b| b == b'\n') {
            let line: Vec<u8> = self.partial.drain(..=end).collect();
            lines.push(String::from_utf8_lossy(&line).trim_end().to_string());
        }
        Ok(lines)
    }
}

async fn follow(path: PathBuf, mut done: oneshot::Receiver<()>) -> std::io::Result<Vec<String>> {
    let mut follower = Follower::new(path);
    let mut seen = Vec::new();
    let mut ticker = interval(Duration::from_millis(10));
    loop {
        tokio::select! {
            _ = ticker.tick() => seen.extend(follower.poll().await?),
            _ = &mut done => {
                seen.extend(follower.poll().await?);
                return Ok(seen);
            }
        }
    }
}

async fn append(path: &Path, text: &str) -> std::io::Result<()> {
    let mut file = OpenOptions::new().create(true).append(true).open(path).await?;
    file.write_all(text.as_bytes()).await?;
    file.flush().await
}

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let dir = std::env::temp_dir().join(format!("tokio-l22-{}-follow", std::process::id()));
    fs::create_dir_all(&dir).await?;
    let log = dir.join("app.log");

    let (done_tx, done_rx) = oneshot::channel();
    let follower = tokio::spawn(follow(log.clone(), done_rx));

    let writer_path = log.clone();
    let writer = tokio::spawn(async move {
        for chunk in ["alpha\n", "be", "ta\n", "gamma\n"] {
            sleep(Duration::from_millis(30)).await;
            append(&writer_path, chunk).await?;
        }
        let _ = done_tx.send(());
        Ok::<_, std::io::Error>(())
    });

    writer.await.expect("writer panicked")?;
    for line in follower.await.expect("follower panicked")? {
        println!("followed: {line}");
    }

    fs::write(&log, "one\ntwo\n").await?;
    let mut check = Follower::new(log.clone());
    println!("first read: {:?}", check.poll().await?);
    fs::write(&log, "three\n").await?;
    println!("after truncation: {:?}", check.poll().await?);
    println!("nothing new: {:?}", check.poll().await?);

    fs::remove_dir_all(&dir).await?;
    Ok(())
}

Output

followed: alpha
followed: beta
followed: gamma
first read: ["one", "two"]
after truncation: ["three"]
nothing new: []

beta appears once, after its second half is written: the follower keeps be in partial and does not print it. The result is the same whether a poll happens between the two writes or not, so the printed lines do not depend on timing. The truncation check replaces the file with a shorter one, and poll reads from the start.

Stage 3: a job runner that follows its own log

The final program runs jobs. run_job starts sh -c script sh LOGFILE, so the script can refer to the log path as $1. It starts a follower on that log, waits for the process under a time limit, and then tells the follower to stop. The follower's last poll runs after the process has exited, so every line the job wrote is collected. Two jobs run at the same time with join!, and the program prints their results in a fixed order after both finish.

Rust

use std::path::PathBuf;
use std::time::Duration;
use tokio::fs;
use tokio::process::Command;
use tokio::sync::oneshot;
use tokio::time::{interval, timeout};

#[derive(Debug)]
enum Outcome {
    Exited(i32),
    Signaled,
    TimedOut,
}

struct Follower {
    path: PathBuf,
    offset: usize,
    partial: Vec<u8>,
}

impl Follower {
    fn new(path: PathBuf) -> Follower {
        Follower { path, offset: 0, partial: Vec::new() }
    }

    async fn poll(&mut self) -> std::io::Result<Vec<String>> {
        let bytes = match fs::read(&self.path).await {
            Ok(bytes) => bytes,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
            Err(e) => return Err(e),
        };
        if bytes.len() < self.offset {
            self.offset = 0;
            self.partial.clear();
        }
        self.partial.extend_from_slice(&bytes[self.offset..]);
        self.offset = bytes.len();

        let mut lines = Vec::new();
        while let Some(end) = self.partial.iter().position(|&b| b == b'\n') {
            let line: Vec<u8> = self.partial.drain(..=end).collect();
            lines.push(String::from_utf8_lossy(&line).trim_end().to_string());
        }
        Ok(lines)
    }
}

async fn follow(path: PathBuf, mut done: oneshot::Receiver<()>) -> std::io::Result<Vec<String>> {
    let mut follower = Follower::new(path);
    let mut seen = Vec::new();
    let mut ticker = interval(Duration::from_millis(10));
    loop {
        tokio::select! {
            _ = ticker.tick() => seen.extend(follower.poll().await?),
            _ = &mut done => {
                seen.extend(follower.poll().await?);
                return Ok(seen);
            }
        }
    }
}

struct JobResult {
    name: String,
    lines: Vec<String>,
    outcome: Outcome,
}

async fn run_job(
    name: &str,
    script: &str,
    limit: Duration,
    dir: &std::path::Path,
) -> std::io::Result<JobResult> {
    let log = dir.join(format!("{name}.log"));
    let mut child = Command::new("sh")
        .arg("-c")
        .arg(script)
        .arg("sh")
        .arg(&log)
        .kill_on_drop(true)
        .spawn()?;

    let (done_tx, done_rx) = oneshot::channel();
    let follower = tokio::spawn(follow(log, done_rx));

    let outcome = match timeout(limit, child.wait()).await {
        Ok(status) => match status?.code() {
            Some(code) => Outcome::Exited(code),
            None => Outcome::Signaled,
        },
        Err(_elapsed) => {
            child.kill().await?;
            Outcome::TimedOut
        }
    };

    let _ = done_tx.send(());
    let lines = follower.await.expect("follower panicked")?;
    Ok(JobResult { name: name.to_string(), lines, outcome })
}

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let dir = std::env::temp_dir().join(format!("tokio-l22-{}-jobs", std::process::id()));
    fs::create_dir_all(&dir).await?;

    let counter = "for i in 1 2 3; do echo \"line $i\" >> \"$1\"; sleep 0.05; done; exit 4";
    let stuck = "echo started >> \"$1\"; exec sleep 30";

    let (first, second) = tokio::join!(
        run_job("counter", counter, Duration::from_secs(10), &dir),
        run_job("stuck", stuck, Duration::from_millis(1000), &dir),
    );

    for result in [first?, second?] {
        for line in &result.lines {
            println!("[{}] {line}", result.name);
        }
        println!("[{}] outcome: {:?}", result.name, result.outcome);
    }

    fs::remove_dir_all(&dir).await?;
    Ok(())
}

Output

[counter] line 1
[counter] line 2
[counter] line 3
[counter] outcome: Exited(4)
[stuck] started
[stuck] outcome: TimedOut

Both jobs run at the same time: the follower tasks poll their logs while the shell scripts run. The stuck job writes one line, then exec sleep 30 replaces the shell with sleep, so the process that holds the timeout is the process that kill ends. The line started is still collected, because the follower polls once more after the kill. Exit code 4 comes from the exit 4 at the end of the first script, which shows that the runner reports the real status.

Part	Role
`Outcome`	how a job ended: exit code, killed by a signal, or timed out
`Follower`	offset and partial-line buffer, `poll` returns the complete new lines
`follow`	polls on an `interval` and does a final poll when told to stop
`run_job`	spawns the script, follows its log, applies the time limit, kills on timeout
`join!` in `main`	runs two jobs concurrently, printing happens after both finish

Extensions:

Capture stdout and stderr into the log file from inside the runner: open the log with OpenOptions, convert it with into_std, and pass it to Command::stdout.
Add a grace step to timeouts: send SIGTERM with the kill command, wait up to two seconds for exit, and only then call Child::kill.
Replace the polling with a rate that adapts: poll every 10 ms while lines keep arriving, and back off to 200 ms after ten empty polls.

Interview points

An echo server shows the standard TCP server shape: bind, an accept loop, one task per connection, and a read loop that ends when read returns 0.
Half-close (shutdown on the write side) lets a client signal "no more input" and still read the rest of the reply.
A client that writes a large payload before it reads can stall against a server that echoes as it reads. Splitting the socket and using a second task for writing avoids this.
A shared store behind Arc<Mutex<_>> stays correct when read-modify-write happens under one guard, and stays cheap when the guard is dropped before any .await.
A chat room combines a per-client select! with a broadcast channel. Both branches are cancel safe, so a lost race loses no data.
Subscribing before announcing a client removes the window in which it could miss a message.
HTTP/1.1 needs framing: the blank line ends the headers and Content-Length gives the body size. A limit on that size protects memory.
Making handlers generic over AsyncRead + AsyncWrite lets duplex replace the network in tests.
A timeout around child.wait() does not stop the child. The code must kill it and reap it.
Reading a child's stdout and stderr one after the other can deadlock. Reading both concurrently avoids it.
A file follower keeps an offset and a partial-line buffer, and handles missing files and truncation.

Summary

Project 10 builds a TCP echo server with TcpListener, one task per connection, half-close with shutdown, into_split for overlapping reads and writes, and a oneshot to stop the accept loop.
Project 11 builds a line-based key-value protocol with BufReader::lines, split, and an Arc<Mutex<HashMap>> locked for one command at a time.
Project 12 builds a chat room from a broadcast channel and a select! per connection, with unique nicknames in a shared set.
Project 13 builds an HTTP/1.1 server by hand: request parsing with limits, routing as a pure function, Content-Length framing, keep-alive, and tests over duplex.
Project 14 builds a process runner with Command, timeout, and kill, and a file follower with fs::read, an offset, and an interval.
Every program binds port 0, drives itself with an in-process client, and prints only lines that do not depend on timing.

Project 10: TCP echo server and client

Stage 1: one connection

Stage 2: an accept loop and many clients

Stage 3: statistics, large messages, and a clean stop

Project 11: Key-value server with a text protocol

Stage 1: parse and execute, with no networking

Stage 2: serve the protocol over TCP

Stage 3: INCR, KEYS, and concurrent clients

Project 12: Chat room server

Stage 1: fan-out of raw lines

Stage 2: nicknames, joins, and leaves

Stage 3: unique nicknames, /users, and lag handling

Project 13: Minimal HTTP/1.1 server

Stage 1: parse a request and write a response

Stage 2: routing and error responses

Stage 3: TCP, keep-alive, and shared state

Project 14: Process runner and file follower

Stage 1: run a command with a time limit

Stage 2: follow a file

Stage 3: a job runner that follows its own log

Interview points

Summary

Stage 3: unique nicknames, `/users`, and lag handling