Workshop 7: Production Practice

Each project starts with a problem and requirements, then grows through stages. Every stage is a complete program that compiles and runs. The last stage is the full program, followed by a table of its parts and extension tasks.

The projects use Testing Async Code (paused time, advance, tokio_test), Error Handling (retry and backoff), Tracing and Observability, Codecs (Framed, LinesCodec), Graceful Shutdown (CancellationToken, TaskTracker). Project 26 is a test suite, so its output is the cargo test report. The Rust blocks are not machine-checked (concept mode); their outputs were derived by reading the code. Project 27 runs under the paused clock, so retry delays cost no real time.

Project 26: Deterministic test suite for retry and rate limiting

Code that retries and throttles is about time, and tests that use real time are slow and flaky. This suite tests two small components by controlling the clock.

Requirements:

retry runs an operation up to max_attempts times and waits base * 2^(attempt - 1) between attempts.
RateLimiter allows one call per gap. Idle time is not saved up.
Tests assert exact elapsed times, not ranges.
Tests check the state between two moments with advance.
Tests check the waker contract with tokio_test::task::spawn.

Stage 1: Retry with exact elapsed time

Under start_paused = true, a sleep completes as soon as the runtime has nothing else to do, and the clock jumps to the deadline. Instant::now() differences are therefore exact.

Rust

use std::future::Future;
use std::time::Duration;
use tokio::time::sleep;

pub async fn retry<T, E, F, Fut>(max_attempts: u32, base: Duration, mut op: F) -> Result<T, E>
where
    F: FnMut(u32) -> Fut,
    Fut: Future<Output = Result<T, E>>,
{
    let mut attempt = 1;
    loop {
        match op(attempt).await {
            Ok(value) => return Ok(value),
            Err(error) if attempt >= max_attempts => return Err(error),
            Err(_) => {
                sleep(base * 2u32.pow(attempt - 1)).await;
                attempt += 1;
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use tokio::time::Instant;

    const BASE: Duration = Duration::from_millis(100);

    #[tokio::test(start_paused = true)]
    async fn first_success_waits_for_nothing() {
        let start = Instant::now();
        let result: Result<u32, u32> = retry(3, BASE, |n| async move { Ok(n) }).await;
        assert_eq!(result, Ok(1));
        assert_eq!(start.elapsed(), Duration::ZERO);
    }

    #[tokio::test(start_paused = true)]
    async fn two_failures_wait_100_then_200() {
        let start = Instant::now();
        let result = retry(3, BASE, |n| async move { if n < 3 { Err(n) } else { Ok(n) } }).await;
        assert_eq!(result, Ok(3));
        assert_eq!(start.elapsed(), Duration::from_millis(300));
    }

    #[tokio::test(start_paused = true)]
    async fn exhausted_retries_return_the_last_error() {
        let start = Instant::now();
        let result: Result<(), u32> = retry(3, BASE, |n| async move { Err(n) }).await;
        assert_eq!(result, Err(3));
        // Two waits: after attempt 1 and after attempt 2. None after the last.
        assert_eq!(start.elapsed(), Duration::from_millis(300));
    }
}

Output

running 3 tests
test tests::exhausted_retries_return_the_last_error ... ok
test tests::first_success_waits_for_nothing ... ok
test tests::two_failures_wait_100_then_200 ... ok

test result: ok. 3 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s

Three attempts mean two waits. The final failure returns at once, so a test that expected a third wait of 400 ms would fail.

Stage 2: A rate limiter and the waker contract

RateLimiter hands out time slots. Each call takes the later of "now" and the next free slot, moves the next slot forward by gap, and sleeps until its own slot. tokio_test::task::spawn wraps a future in a mock task that the test polls by hand. is_woken shows whether the timer woke the task. The tests poll acquire and retry directly and use advance to step the clock. The block is the full file.

Rust

use std::future::Future;
use std::sync::Mutex;
use std::time::Duration;
use tokio::time::{sleep, sleep_until, Instant};

pub async fn retry<T, E, F, Fut>(max_attempts: u32, base: Duration, mut op: F) -> Result<T, E>
where
    F: FnMut(u32) -> Fut,
    Fut: Future<Output = Result<T, E>>,
{
    let mut attempt = 1;
    loop {
        match op(attempt).await {
            Ok(value) => return Ok(value),
            Err(error) if attempt >= max_attempts => return Err(error),
            Err(_) => {
                sleep(base * 2u32.pow(attempt - 1)).await;
                attempt += 1;
            }
        }
    }
}

pub struct RateLimiter {
    gap: Duration,
    next: Mutex<Instant>,
}

impl RateLimiter {
    pub fn new(gap: Duration) -> Self {
        RateLimiter { gap, next: Mutex::new(Instant::now()) }
    }

    pub async fn acquire(&self) {
        let slot = {
            let mut next = self.next.lock().unwrap();
            let slot = (*next).max(Instant::now());
            *next = slot + self.gap;
            slot
        };
        sleep_until(slot).await;
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use tokio::time::advance;

    const GAP: Duration = Duration::from_millis(100);

    #[tokio::test(start_paused = true)]
    async fn sequential_calls_are_spaced() {
        let limiter = RateLimiter::new(GAP);
        let start = Instant::now();
        let mut times = Vec::new();
        for _ in 0..3 {
            limiter.acquire().await;
            times.push(start.elapsed().as_millis());
        }
        assert_eq!(times, [0, 100, 200]);
    }

    #[tokio::test(start_paused = true)]
    async fn acquire_wakes_its_task_at_the_slot() {
        use tokio_test::{assert_pending, assert_ready, task};

        let limiter = RateLimiter::new(GAP);
        assert_ready!(task::spawn(limiter.acquire()).poll()); // first call: no wait

        let mut second = task::spawn(limiter.acquire());
        assert_pending!(second.poll());
        advance(Duration::from_millis(99)).await;
        assert!(!second.is_woken());
        advance(Duration::from_millis(1)).await;
        assert!(second.is_woken());
        assert_ready!(second.poll());
    }

    #[tokio::test(start_paused = true)]
    async fn retry_sleeps_between_attempts() {
        use tokio_test::{assert_pending, assert_ready_eq, task};

        let mut run = task::spawn(retry(3, Duration::from_millis(100), |n| async move {
            if n < 3 { Err(n) } else { Ok(n) }
        }));
        assert_pending!(run.poll()); // attempt 1 failed, sleeping 100 ms
        advance(Duration::from_millis(100)).await;
        assert!(run.is_woken());
        assert_pending!(run.poll()); // attempt 2 failed, sleeping 200 ms
        advance(Duration::from_millis(199)).await;
        assert!(!run.is_woken());
        advance(Duration::from_millis(1)).await;
        assert_ready_eq!(run.poll(), Ok(3));
    }
}

Output

running 3 tests
test tests::acquire_wakes_its_task_at_the_slot ... ok
test tests::retry_sleeps_between_attempts ... ok
test tests::sequential_calls_are_spaced ... ok

test result: ok. 3 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s

The final poll of retry returns Ok(3) only after the test moved the clock by exactly 100 ms and then 200 ms.

Parts of the suite:

Part	Role
`start_paused = true`	the clock moves only when the runtime is idle, so elapsed times are exact
`Instant::elapsed`	measures virtual time in assertions
`advance`	steps the clock to check state at a boundary
`task::spawn`, `is_woken`, `assert_pending!`	poll by hand and check the waker contract

Extension tasks:

Add a max_delay cap to retry and test that the waits are 100, 200, 400, 400.
Make the limiter allow bursts of n calls and test the first n complete at time 0.
Add a failing test with test_fail for a limiter that saves idle time, to show which assertion catches it.

Project 27: Capstone job-queue service

A service accepts jobs over TCP, runs them on a worker, retries failed jobs with backoff, moves jobs that keep failing to a dead-letter queue (a list of jobs given up on), keeps counters, logs with tracing, and shuts down without losing accepted jobs.

Requirements:

The protocol is one text line per message. Commands: SUBMIT <name> <fails>, STATS, DEAD. A job's fails is how many leading attempts fail (a simulation of a flaky task).
A codec turns lines into typed commands and typed replies.
A job gets at most 3 attempts, with waits of 100 ms and 200 ms between them. A third failure sends the job to the dead-letter queue.
Counters: submitted, done, retried, dead, pending.
Log events go to a buffer, and the program prints them at the end, so the order is fixed.
Shutdown stops accepting, ends connections, and lets the worker finish every queued job before the program ends.
An in-process client on port 0 drives the run.

Stage 1: Codec, retries, backoff, and the dead-letter queue

The codec wraps LinesCodec and adds the typed layer. A parse failure becomes a Command::Invalid value, so a bad line gets an error reply and the connection stays open. The worker runs each job through up to MAX_ATTEMPTS attempts. An attempt n fails when n <= job.fails. After a failed attempt that is not the last, the worker waits 100 ms * 2^(n - 1). A job that fails every attempt goes to the dead list. STATS reports submitted, done, retried, dead, and pending counts. DEAD lists the names. The client waits until pending=0 by polling STATS.

Rust

use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;

use futures::{SinkExt, StreamExt};
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::mpsc;
use tokio::time::sleep;
use tokio_util::bytes::BytesMut;
use tokio_util::codec::{Decoder, Encoder, Framed, LinesCodec, LinesCodecError};

const MAX_ATTEMPTS: u32 = 3;
const BASE_DELAY: Duration = Duration::from_millis(100);

enum Command {
    Submit { name: String, fails: u32 },
    Stats,
    Dead,
    Invalid(String),
}

enum Reply {
    Accepted(u64),
    Text(String),
    Error(String),
}

struct QueueCodec(LinesCodec);

impl Decoder for QueueCodec {
    type Item = Command;
    type Error = LinesCodecError;

    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Command>, LinesCodecError> {
        let Some(line) = self.0.decode(src)? else { return Ok(None) };
        let parts: Vec<&str> = line.split_whitespace().collect();
        let command = match parts.as_slice() {
            ["SUBMIT", name, fails] => match fails.parse() {
                Ok(fails) => Command::Submit { name: name.to_string(), fails },
                Err(_) => Command::Invalid(format!("bad fail count {fails}")),
            },
            ["STATS"] => Command::Stats,
            ["DEAD"] => Command::Dead,
            _ => Command::Invalid(format!("unknown command {line}")),
        };
        Ok(Some(command))
    }
}

impl Encoder<Reply> for QueueCodec {
    type Error = LinesCodecError;

    fn encode(&mut self, reply: Reply, dst: &mut BytesMut) -> Result<(), LinesCodecError> {
        let line = match reply {
            Reply::Accepted(id) => format!("OK {id}"),
            Reply::Text(text) => text,
            Reply::Error(message) => format!("ERR {message}"),
        };
        self.0.encode(line, dst)
    }
}

struct Job {
    name: String,
    fails: u32,
}

#[derive(Default)]
struct Shared {
    submitted: AtomicU64,
    done: AtomicU64,
    retried: AtomicU64,
    dead: Mutex<Vec<String>>,
}

impl Shared {
    fn stats(&self) -> String {
        let submitted = self.submitted.load(Ordering::SeqCst);
        let done = self.done.load(Ordering::SeqCst);
        let dead = self.dead.lock().unwrap().len() as u64;
        format!(
            "STATS submitted={submitted} done={done} retried={} dead={dead} pending={}",
            self.retried.load(Ordering::SeqCst),
            submitted - done - dead,
        )
    }
}

async fn run_job(job: &Job, shared: &Shared) {
    for attempt in 1..=MAX_ATTEMPTS {
        if attempt > job.fails {
            shared.done.fetch_add(1, Ordering::SeqCst);
            return;
        }
        if attempt < MAX_ATTEMPTS {
            shared.retried.fetch_add(1, Ordering::SeqCst);
            sleep(BASE_DELAY * 2u32.pow(attempt - 1)).await;
        }
    }
    shared.dead.lock().unwrap().push(job.name.clone());
}

async fn worker(mut jobs: mpsc::Receiver<Job>, shared: Arc<Shared>) {
    while let Some(job) = jobs.recv().await {
        run_job(&job, &shared).await;
    }
}

async fn handle(stream: TcpStream, jobs: mpsc::Sender<Job>, shared: Arc<Shared>) {
    let mut framed = Framed::new(stream, QueueCodec(LinesCodec::new()));
    while let Some(Ok(command)) = framed.next().await {
        let reply = match command {
            Command::Submit { name, fails } => {
                let id = shared.submitted.fetch_add(1, Ordering::SeqCst) + 1;
                jobs.send(Job { name, fails }).await.unwrap();
                Reply::Accepted(id)
            }
            Command::Stats => Reply::Text(shared.stats()),
            Command::Dead => Reply::Text(format!("DEAD {}", shared.dead.lock().unwrap().join(","))),
            Command::Invalid(message) => Reply::Error(message),
        };
        if framed.send(reply).await.is_err() {
            break;
        }
    }
}

async fn ask(client: &mut Framed<TcpStream, LinesCodec>, line: &str) -> String {
    client.send(line.to_string()).await.unwrap();
    client.next().await.unwrap().unwrap()
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let shared = Arc::new(Shared::default());
    let (tx, rx) = mpsc::channel(16);
    tokio::spawn(worker(rx, shared.clone()));

    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let server_shared = shared.clone();
    tokio::spawn(async move {
        loop {
            let (stream, _) = listener.accept().await.unwrap();
            tokio::spawn(handle(stream, tx.clone(), server_shared.clone()));
        }
    });

    let stream = TcpStream::connect(addr).await.unwrap();
    let mut client = Framed::new(stream, LinesCodec::new());
    let start = tokio::time::Instant::now();
    for line in ["SUBMIT resize 0", "SUBMIT upload 2", "SUBMIT export 5", "SUBMIT sync x"] {
        println!("{line:<16} -> {}", ask(&mut client, line).await);
    }
    loop {
        let stats = ask(&mut client, "STATS").await;
        if stats.ends_with("pending=0") {
            println!("{stats}");
            break;
        }
        sleep(Duration::from_millis(50)).await;
    }
    println!("{}", ask(&mut client, "DEAD").await);
    println!("virtual time: {} ms", start.elapsed().as_millis());
}

Output

SUBMIT resize 0  -> OK 1
SUBMIT upload 2  -> OK 2
SUBMIT export 5  -> OK 3
SUBMIT sync x    -> ERR bad fail count x
STATS submitted=3 done=2 retried=4 dead=1 pending=0
DEAD export
virtual time: 626 ms

upload fails twice, so it waits 100 ms and 200 ms and succeeds on attempt 3. export fails every attempt, so it waits the same 300 ms and lands in the dead-letter queue. The virtual time is fixed because the clock is paused: it is the worker's waits plus the client's polling sleeps.

Stage 2: Tracing and graceful shutdown

Log events go into a shared buffer through a custom writer, and the program prints the buffer at the end. Each job runs inside a span with its id and name. Shutdown uses the three steps from Graceful Shutdown: cancel a CancellationToken, wait for the TaskTracker that holds the connection tasks, then wait for the worker. The worker ends when every Sender is dropped, and it drains the queue first. The client submits a last job and starts shutdown at once, to show that the accepted job still finishes.

Rust

use std::io::Write;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;

use futures::{SinkExt, StreamExt};
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::mpsc;
use tokio::time::sleep;
use tokio_util::bytes::BytesMut;
use tokio_util::codec::{Decoder, Encoder, Framed, LinesCodec, LinesCodecError};
use tokio_util::sync::CancellationToken;
use tokio_util::task::TaskTracker;
use tracing::{info, instrument, warn};

const MAX_ATTEMPTS: u32 = 3;
const BASE_DELAY: Duration = Duration::from_millis(100);

enum Command {
    Submit { name: String, fails: u32 },
    Stats,
    Dead,
    Invalid(String),
}

enum Reply {
    Accepted(u64),
    Text(String),
    Error(String),
}

struct QueueCodec(LinesCodec);

impl Decoder for QueueCodec {
    type Item = Command;
    type Error = LinesCodecError;

    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Command>, LinesCodecError> {
        let Some(line) = self.0.decode(src)? else { return Ok(None) };
        let parts: Vec<&str> = line.split_whitespace().collect();
        let command = match parts.as_slice() {
            ["SUBMIT", name, fails] => match fails.parse() {
                Ok(fails) => Command::Submit { name: name.to_string(), fails },
                Err(_) => Command::Invalid(format!("bad fail count {fails}")),
            },
            ["STATS"] => Command::Stats,
            ["DEAD"] => Command::Dead,
            _ => Command::Invalid(format!("unknown command {line}")),
        };
        Ok(Some(command))
    }
}

impl Encoder<Reply> for QueueCodec {
    type Error = LinesCodecError;

    fn encode(&mut self, reply: Reply, dst: &mut BytesMut) -> Result<(), LinesCodecError> {
        let line = match reply {
            Reply::Accepted(id) => format!("OK {id}"),
            Reply::Text(text) => text,
            Reply::Error(message) => format!("ERR {message}"),
        };
        self.0.encode(line, dst)
    }
}

struct Job {
    id: u64,
    name: String,
    fails: u32,
}

#[derive(Default)]
struct Shared {
    submitted: AtomicU64,
    done: AtomicU64,
    retried: AtomicU64,
    dead: Mutex<Vec<String>>,
}

impl Shared {
    fn stats(&self) -> String {
        let submitted = self.submitted.load(Ordering::SeqCst);
        let done = self.done.load(Ordering::SeqCst);
        let dead = self.dead.lock().unwrap().len() as u64;
        format!(
            "STATS submitted={submitted} done={done} retried={} dead={dead} pending={}",
            self.retried.load(Ordering::SeqCst),
            submitted - done - dead,
        )
    }
}

#[instrument(skip_all, fields(id = job.id, name = %job.name))]
async fn run_job(job: &Job, shared: &Shared) {
    for attempt in 1..=MAX_ATTEMPTS {
        if attempt > job.fails {
            shared.done.fetch_add(1, Ordering::SeqCst);
            info!(attempt, "done");
            return;
        }
        if attempt < MAX_ATTEMPTS {
            shared.retried.fetch_add(1, Ordering::SeqCst);
            let delay = BASE_DELAY * 2u32.pow(attempt - 1);
            warn!(attempt, delay_ms = delay.as_millis() as u64, "attempt failed, retrying");
            sleep(delay).await;
        }
    }
    warn!("attempts exhausted, moved to the dead-letter queue");
    shared.dead.lock().unwrap().push(job.name.clone());
}

async fn worker(mut jobs: mpsc::Receiver<Job>, shared: Arc<Shared>) {
    while let Some(job) = jobs.recv().await {
        run_job(&job, &shared).await;
    }
    info!("queue closed, worker stopped");
}

async fn handle(
    stream: TcpStream,
    jobs: mpsc::Sender<Job>,
    shared: Arc<Shared>,
    token: CancellationToken,
) {
    let mut framed = Framed::new(stream, QueueCodec(LinesCodec::new()));
    loop {
        let command = tokio::select! {
            _ = token.cancelled() => break,
            frame = framed.next() => match frame {
                Some(Ok(command)) => command,
                _ => break,
            },
        };
        let reply = match command {
            Command::Submit { name, fails } => {
                let id = shared.submitted.fetch_add(1, Ordering::SeqCst) + 1;
                jobs.send(Job { id, name, fails }).await.unwrap();
                Reply::Accepted(id)
            }
            Command::Stats => Reply::Text(shared.stats()),
            Command::Dead => Reply::Text(format!("DEAD {}", shared.dead.lock().unwrap().join(","))),
            Command::Invalid(message) => Reply::Error(message),
        };
        if framed.send(reply).await.is_err() {
            break;
        }
    }
}

async fn serve(
    listener: TcpListener,
    jobs: mpsc::Sender<Job>,
    shared: Arc<Shared>,
    token: CancellationToken,
    connections: TaskTracker,
) {
    loop {
        tokio::select! {
            _ = token.cancelled() => break,
            accepted = listener.accept() => {
                let (stream, _) = accepted.unwrap();
                connections.spawn(handle(stream, jobs.clone(), shared.clone(), token.clone()));
            }
        }
    }
}

#[derive(Clone)]
struct LogBuffer(Arc<Mutex<Vec<u8>>>);

impl Write for LogBuffer {
    fn write(&mut self, data: &[u8]) -> std::io::Result<usize> {
        self.0.lock().unwrap().extend_from_slice(data);
        Ok(data.len())
    }
    fn flush(&mut self) -> std::io::Result<()> {
        Ok(())
    }
}

async fn ask(client: &mut Framed<TcpStream, LinesCodec>, line: &str) -> String {
    client.send(line.to_string()).await.unwrap();
    client.next().await.unwrap().unwrap()
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let log = LogBuffer(Arc::new(Mutex::new(Vec::new())));
    let writer = log.clone();
    tracing_subscriber::fmt()
        .with_writer(move || writer.clone())
        .with_ansi(false)
        .without_time()
        .with_target(false)
        .init();

    let shared = Arc::new(Shared::default());
    let token = CancellationToken::new();
    let connections = TaskTracker::new();
    let (tx, rx) = mpsc::channel(16);
    let worker_task = tokio::spawn(worker(rx, shared.clone()));

    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let server = tokio::spawn(serve(listener, tx, shared.clone(), token.clone(), connections.clone()));

    let stream = TcpStream::connect(addr).await.unwrap();
    let mut client = Framed::new(stream, LinesCodec::new());
    for line in ["SUBMIT resize 0", "SUBMIT upload 2", "SUBMIT export 5", "SUBMIT sync x"] {
        println!("{line:<16} -> {}", ask(&mut client, line).await);
    }
    loop {
        let stats = ask(&mut client, "STATS").await;
        if stats.ends_with("pending=0") {
            println!("{stats}");
            break;
        }
        sleep(Duration::from_millis(50)).await;
    }
    println!("{}", ask(&mut client, "DEAD").await);

    println!("SUBMIT last 1    -> {}", ask(&mut client, "SUBMIT last 1").await);
    println!("shutting down");
    token.cancel();
    server.await.unwrap();
    connections.close();
    connections.wait().await;
    worker_task.await.unwrap();

    println!("{}", shared.stats());
    println!("--- log ---");
    print!("{}", String::from_utf8(log.0.lock().unwrap().clone()).unwrap());
}

Output

SUBMIT resize 0  -> OK 1
SUBMIT upload 2  -> OK 2
SUBMIT export 5  -> OK 3
SUBMIT sync x    -> ERR bad fail count x
STATS submitted=3 done=2 retried=4 dead=1 pending=0
DEAD export
SUBMIT last 1    -> OK 4
shutting down
STATS submitted=4 done=3 retried=5 dead=1 pending=0
--- log ---
 INFO run_job{id=1 name=resize}: done attempt=1
 WARN run_job{id=2 name=upload}: attempt failed, retrying attempt=1 delay_ms=100
 WARN run_job{id=2 name=upload}: attempt failed, retrying attempt=2 delay_ms=200
 INFO run_job{id=2 name=upload}: done attempt=3
 WARN run_job{id=3 name=export}: attempt failed, retrying attempt=1 delay_ms=100
 WARN run_job{id=3 name=export}: attempt failed, retrying attempt=2 delay_ms=200
 WARN run_job{id=3 name=export}: attempts exhausted, moved to the dead-letter queue
 WARN run_job{id=4 name=last}: attempt failed, retrying attempt=1 delay_ms=100
 INFO run_job{id=4 name=last}: done attempt=2
 INFO queue closed, worker stopped

last 1 was accepted immediately before token.cancel(). The connection task ends, the server drops its Sender, and the worker still runs last (one retry, then done) before it sees the closed queue and stops. The counters after shutdown include that job.

Parts of the service:

Part	Role
`QueueCodec`	`Decoder` and `Encoder` over `LinesCodec`; text lines to `Command` and `Reply`
`Shared`	atomic counters and the dead-letter list
`worker`, `run_job`	one queue consumer; attempts, backoff, dead-lettering, one span per job
`handle`, `serve`	per-connection loop and accept loop, both stopped by the token
`TaskTracker`	lets shutdown wait for connection tasks
`LogBuffer`	a writer that collects log lines so the program prints them in a fixed place

Extension tasks:

Add a SHUTDOWN command that cancels the token from inside a connection.
Run two workers that share the queue behind a Semaphore-bounded pool, and print only totals so the output stays fixed.
Add a RETRY <name> command that moves a job from the dead-letter queue back to the main queue.

Interview points

Under start_paused = true, virtual time advances only when the runtime is idle, so elapsed-time assertions can be exact.
advance checks the state between two moments. Yield after it so other tasks run.
tokio_test::task::spawn with is_woken tests the waker contract without an executor thread.
A retry with n attempts has n - 1 waits.
A time-slot rate limiter takes max(now, next_slot) and does not save idle time.
A codec keeps parsing out of connection handlers and turns bad input into a value instead of an error that closes the stream.
Graceful shutdown order: stop accepting, end connections, wait for the tracker, drop every Sender, then wait for the consumer to drain.
A bounded queue gives submitters backpressure when the worker is slow.

Summary

Deterministic tests use start_paused, Instant::elapsed, advance, and tokio_test.
Project 26 tests retry and RateLimiter with exact millisecond assertions.
Project 27 combines Framed with a custom codec, retries with exponential backoff, a dead-letter queue, atomic counters, tracing spans, and drain-on-shutdown.
The service runs on 127.0.0.1:0 with an in-process client, so its output is fixed.