Hedronite · Backend Dev · Rust · Thu 2026-10-22

Background jobs a bounded queue, workers, and a clean drain in axum on Tokio

Record the job, answer 202, refuse with 503 when the queue is full, and finish what you accepted before you exit.

Lesson Class: Dev (Rust backend · background jobs)
Tested with: axum 0.8.9 · tokio 1.54.1 · rustc/cargo 1.99.0
Lab: crate/ · cargo test (9)
Source: RFC 9110 · tokio mpsc, Runtime and JoinSet docs · axum 0.8.9 serve docs
Respond Now
202 with a Location and a status record; a bare spawn is dropped at shutdown.
Bounded Queue
try_send answers at once: Full becomes 503 with Retry-After, Closed means stopping.
Drain, Then Exit
Drop every Sender, let recv return None after the buffer empties, await the JoinSet.
Finish what you accepted.

The problem

Suppose your API has an endpoint that renders a report, and the render takes several seconds. If the handler does the work itself, the client waits the whole time with its connection held open. A slow client, a proxy with a short timeout, or a burst of requests can then pin down every connection you have, and every one of them is waiting on the same slow task.

The obvious fix is to call tokio::spawn inside the handler and return at once. That response is fast, but the approach has three faults.

  1. Nothing limits how much work piles up. Each request adds another task, and the process keeps accepting work with no ceiling.
  2. The client has no record to check. It got "OK" and has no id, no status, and no way to learn whether the report finished or failed.
  3. The work is silently dropped at shutdown. A spawned task belongs to the runtime, and nobody waits for it. When the process stops, tasks that have not finished simply vanish, with no error and no log line.

The service below fixes all three with Tokio and axum alone: a bounded queue, a fixed set of worker tasks, a status endpoint, and a shutdown that finishes what was accepted. Every number quoted comes from a nine-test lab run with Tokio 1.54.1 and axum 0.8.9.

The idea, in plain words

When a request arrives, record the job in a map with the status "queued", then put it on a bounded queue. Answer 202 Accepted with a Location header that says where to look for the result. A fixed number of worker tasks take jobs off the queue one at a time and update the record as they go, retrying a failed job up to three attempts. If the queue is full, the handler refuses with 503 and a Retry-After header instead of piling up more work. At shutdown, the service stops accepting new jobs, lets the workers finish everything already queued, and only then exits.

Respond now, work later

A 202 Accepted response says the server took the request but has not finished it. RFC 9110 puts it this way: "the request has been accepted for processing, but the processing has not been completed." It also says what the body should contain: the representation "ought to describe the request's current status and point to (or embed) a status monitor." In this service the status monitor is GET /jobs/{id}, and the status record is an entry in a shared map.

The reason to do it this way is that the client's connection is released immediately, and the client keeps a handle it can poll. Without the record, "accepted" is a promise the client cannot check.

The status is an enum, and the response wraps it with the id:

#[derive(Clone, Debug, PartialEq, Serialize)]
#[serde(tag = "status", rename_all = "lowercase")]
pub enum JobStatus {
    Queued,
    Running { attempt: u32 },
    Done { output: String, attempts: u32 },
    Failed { error: String, attempts: u32 },
}

#[derive(Serialize)]
pub struct JobView {
    pub id: u64,
    #[serde(flatten)]
    pub status: JobStatus,
}

tag = "status" makes serde write the variant name into a status field next to the variant's own fields, and flatten on JobView merges those fields into the same JSON object as id. The result is flat: {"id":1,"status":"done","output":"HELLO","attempts":1}. A client reads one object and switches on status, with no nesting to dig through. The variants carry only what is meaningful in that state, so a Running job has an attempt number and no output yet.

Now the handler that creates the record:

async fn enqueue(State(s): State<AppState>, Json(req): Json<NewJob>) -> Response {
    let id = s.next_id.fetch_add(1, Ordering::Relaxed) + 1;
    // Record the job before a worker can see it, so a fast worker never
    // updates an id the map does not know yet.
    set(&s.jobs, id, JobStatus::Queued);
    let job = Job { id, input: req.input, work_ms: req.work_ms };
    match s.tx.try_send(job) {
        Ok(()) => (
            StatusCode::ACCEPTED,
            [(header::LOCATION, format!("/jobs/{id}"))],
            Json(JobView { id, status: JobStatus::Queued }),
        )
            .into_response(),
        // Err(TrySendError::Full(..)) and Err(TrySendError::Closed(..))
        // are shown in the next section.
    }
}

The id comes from an AtomicU64 shared through AppState, so concurrent requests never get the same number. The important ordering is that set(..., Queued) runs before try_send. Once the job is on the channel a worker may pick it up instantly and write Running into the map, and if the Queued insert came later it would overwrite that newer state, or the worker would update an id the map does not know yet. On success the handler returns 202, a Location of /jobs/1, and a body showing the queued status.

Lab test t1 confirms this: 202 location=Some("/jobs/1") body={"id":1,"status":"queued"}. In t2, the first GET arrived after a worker had taken the job and returned {"id":1,"status":"running","attempt":1}; after the job settled, the last GET returned 200 {"id":1,"status":"done","output":"HELLO","attempts":1}. In t4, a GET /jobs/999 for an id that was never issued returned 404 no job with id 999, so the client can tell "unknown" from "not finished".

The retry rule lives in the worker. Lab test t5 submits a job whose input is fail, which always errors. The worker log shows job 1 attempt 1 failed: input rejected: fail and the same for attempt 2, and the final status is {"id":1,"status":"failed","error":"input rejected: fail","attempts":3}. The retries here are immediate, with no delay between attempts. Backoff is not built.

Now the approach this replaces. Test t7 spawns five bare tasks that each sleep 50 ms and then count themselves done, and drops the runtime:

rt.block_on(async {
    for _ in 0..5 {
        let d = done.clone();
        tokio::spawn(async move {
            tokio::time::sleep(Duration::from_millis(50)).await;
            d.fetch_add(1, SeqCst);
        });
    }
});
drop(rt);

block_on returns as soon as the loop has spawned the five tasks, long before any 50 ms sleep ends. Dropping rt then shuts the runtime down. The output was t7 fire-and-forget completed 0/5 after runtime drop. The Tokio docs describe this behavior: "Tasks spawned through Runtime::spawn keep running until they yield. Then they are dropped. They are not guaranteed to run to completion, but might do so if they do not yield until completion." A sleeping task yields right away, so it is dropped.

Common mistake: bare spawn in the handler

A bounded queue is the backpressure

A bounded channel holds at most a fixed number of messages. Tokio's mpsc::channel is documented as "a bounded mpsc channel for communicating between asynchronous tasks with backpressure", and "once the buffer is full, attempts to send new messages will wait until a message is received from the channel." The word backpressure means that a slow consumer pushes back on the producer instead of letting work accumulate without limit. That is the fix for the first fault: the queue capacity is the cap on waiting work.

There is a catch. If the handler used send().await, a full queue would make the HTTP request wait for space, which brings back the connection-holding problem from the start. So the handler uses try_send, which per the docs differs from send "by returning immediately if the channel's buffer is full or no receiver is waiting". It also has "two failure cases instead of one (one for disconnection, one for a full buffer)". Those are TrySendError::Full and TrySendError::Closed, and the handler maps both to 503:

Err(TrySendError::Full(job)) => {
    s.jobs.lock().unwrap().remove(&job.id);
    (StatusCode::SERVICE_UNAVAILABLE, [(header::RETRY_AFTER, "1")], "queue full, retry later")
        .into_response()
}
Err(TrySendError::Closed(job)) => {
    s.jobs.lock().unwrap().remove(&job.id);
    (StatusCode::SERVICE_UNAVAILABLE, "shutting down").into_response()
}

Both errors hand the job back inside the error value ("The error includes the value passed to send"), and the handler uses job.id to remove the Queued entry it inserted a moment earlier. Otherwise a refused job would sit in the map as "queued" forever, and a client holding no Location could never ask about it anyway. The two arms differ in what they tell the client. A full queue is a temporary overload, so it gets Retry-After: 1. RFC 9110 describes 503 as "currently unable to handle the request due to a temporary overload or scheduled maintenance" and says the server "MAY send a Retry-After header field" to suggest a wait. A closed queue means the service is stopping, so the body says "shutting down" and no retry hint is sent.

Test t3 shows the full case. The queue has capacity 2 and one worker. Each job takes 500 ms, and the test sleeps 20 ms after each POST so the worker has time to take job 1 off the buffer. Job 1 occupies the worker, jobs 2 and 3 fill the buffer, and the fourth POST is refused: 503 retry-after=Some("1") body=queue full, retry later, with codes [202, 202, 202, 503].

Several workers share the queue even though a channel supports only one consumer: "Only one Receiver is supported." So start puts the single Receiver behind an Arc<tokio::sync::Mutex<...>> and gives each worker a clone of the Arc:

let (tx, rx) = mpsc::channel::<Job>(capacity);
let rx = Arc::new(tokio::sync::Mutex::new(rx));
// ... in each worker:
let next = { rx.lock().await.recv().await };
let Some(job) = next else { break }; // closed and empty

Only one worker holds the lock and waits on recv at a time. The braces end the lock's scope as soon as recv returns a job or None, so the next idle worker can take the lock while the first one runs its job. Taking a job is serialized, but running jobs is parallel. Test t8 shows it: 4 jobs of 100 ms on 2 workers settled in about 214 ms, where one worker doing them in sequence would need about 400 ms.

Common mistake: a queue with zero workers

Drain on shutdown

Draining means refusing new work while finishing the work already accepted. It depends on how a channel closes. Receiver::recv "returns None if the channel has been closed and there are no remaining messages in the channel's buffer." and "the channel is closed when all senders have been dropped, or when close is called." The order matters: None comes only after the buffer is empty, so closing the channel does not throw away queued jobs. Each worker keeps receiving until the buffer runs dry and then sees None.

The workers also need an owner that waits for them. JoinSet is Tokio's collection of tasks, and its docs warn: "When the JoinSet is dropped, all tasks in the JoinSet are immediately aborted." So dropping the set would cut off a job in the middle of its work, the same loss as the bare spawn. Shutdown has to await the set instead:

pub async fn shutdown(self) -> usize {
    let Queue { state, mut workers } = self;
    drop(state);
    let mut stopped = 0;
    while let Some(res) = workers.join_next().await {
        res.expect("worker panicked");
        stopped += 1;
    }
    stopped
}

drop(state) removes the Sender stored in the queue. If no other Sender clone exists, the channel is now closed. Each worker finishes its current job, keeps receiving until the buffer is empty, gets None, breaks out of its loop, logs queue closed, exiting, and returns. join_next yields one finished worker at a time, and the loop ends when the set is empty. The return value is the number of workers that stopped, which a caller can log or check.

Test t6 queued 6 jobs of 40 ms on 2 workers, dropped the router, and called shutdown. It printed shutdown drained: 6/6 done, 2 workers stopped, in 127.176042ms, and both workers logged queue closed, exiting. Every accepted job finished, so the third fault is fixed. The timing checks out against the arithmetic: six 40 ms jobs split across two workers is three rounds, or 120 ms of sleeping, and the remaining 7 ms is scheduling and polling overhead.

In the binary, the order is wired in main.rs:

#[tokio::main]
async fn main() {
    let queue = jobs_lab::start(64, 4);
    let app = queue.router();
    let listener = tokio::net::TcpListener::bind("127.0.0.1:3000").await.unwrap();
    axum::serve(listener, app)
        .with_graceful_shutdown(async { tokio::signal::ctrl_c().await.unwrap() })
        .await
        .unwrap();
    // The server has returned and dropped its router, so the last Sender is
    // the one inside `queue`. shutdown() drops it and drains the workers.
    let stopped = queue.shutdown().await;
    eprintln!("drained; {stopped} workers stopped");
}

queue.router() clones the state, so the Router holds its own Sender. with_graceful_shutdown prepares the server to stop when the signal future completes, and axum documents that the serve future "returns Ok(()) only after the signal future completes." By the time .await returns, the server has dropped its Router, and that Sender clone is gone. Only then does main call queue.shutdown(), which drops the last Sender and drains. One limit on this claim: the Ctrl-C path in main was compiled but not exercised by a test. The tests drive shutdown directly.

Common mistake: a live Sender clone

Putting it together

The nine tests, mapped to the three faults from the problem:

TestWhat it provedFault it answers
t1POST returns 202 with Location: /jobs/1 and a queued bodyNo record to check
t2Status moves from running to done, and the JSON is flatNo record to check
t3Full queue gives 503 with Retry-After: 1; codes [202, 202, 202, 503]No limit on pile-up
t4Unknown id gives 404No record to check
t5A failing job retries and ends failed with 3 attemptsNo record to check
t6Shutdown drains 6/6 queued jobs and stops 2 workersWork lost at shutdown
t7Bare spawned tasks complete 0/5 after the runtime dropsWork lost at shutdown
t82 workers settle 4 jobs of 100 ms in about 214 msWorkers run in parallel
t9A live Sender clone blocks shutdown (timeout)Work lost at shutdown

How to run it

Dependencies in Cargo.toml:

[dependencies]
axum = "=0.8.9"
serde = { version = "1", features = ["derive"] }
serde_json = "1"
tokio = { version = "1", features = ["full"] }

[dev-dependencies]
tower = { version = "0.5", features = ["util"] }
http-body-util = "0.1"

Run the tests with:

cargo test --test jobs -- --nocapture --test-threads=1

The real result line from the lab run:

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

Tested with rustc/cargo 1.99.0, axum 0.8.9, and tokio 1.54.1 (the tokio = "1" requirement resolved to 1.54.1 in Cargo.lock), on hyper 1.12.0 and tower 0.5.3.

Limits

Check yourself

  1. Why does enqueue insert Queued into the map before calling try_send? *A fast worker could take the job and write Running before the insert, and the Queued write would then overwrite it, or the worker would update an id the map does not know yet.*
  2. Why use try_send instead of send().await in the handler? *send waits for buffer space, which makes the HTTP request wait. try_send returns at once with Full or Closed, so the handler can answer 503.*
  3. Several workers share one Receiver. How do jobs still run in parallel? *The Receiver sits behind a tokio Mutex, and the lock is released as soon as recv returns a job, so only the take is serialized. In t8, 4 jobs of 100 ms on 2 workers settled in about 214 ms.*
  4. When does recv return None, and why does that make draining safe? Only when the channel is closed and the buffer is empty. Queued jobs are still delivered first, so workers exit after finishing everything accepted.
  5. Shutdown hangs and a 300 ms timeout fires. What is the likely cause? *A Sender clone is still alive, such as a Router that was not dropped, so the channel never closes. Drop every clone before calling shutdown.*

Sources