Bài 18: Async Handler Và tokio Runtime

Sau bài học, bạn sẽ:

• Hiểu sâu lý do handler axum BẮT BUỘC async fn (không sync) — axum trên tokio multi-thread executor, một worker thread phục vụ nhiều request concurrent qua await yield.
• Nắm macro #[tokio::main] làm gì — desugar thành tokio::runtime::Builder::new_multi_thread().enable_all().build()?.block_on(async { ... }) với multi_thread flavor mặc định, worker_threads tự động theo số CPU.
• Hiểu trait bound Send + 'static cho Future trả về từ handler — compiler enforce qua trait Handler của axum (đã preview B13).
• Biết khi nào dùng tokio::spawn trong handler — fire-and-forget background task nhẹ như send email confirmation, audit log async, cache invalidation, webhook delivery retry.
• Biết khác biệt tokio::time::sleep(Duration).await async yield vs std::thread::sleep(Duration) blocking worker.
• Hiểu pitfall blocking operation trong handler async (CPU-heavy, file IO sync, blocking HTTP client) và cách workaround qua tokio::task::spawn_blocking.
• Áp dụng vào Shop API: pattern fire-and-forget cho audit log + email confirmation ban đầu, G21 migrate sang Redis job queue (apalis) cho task quan trọng cần track.

axum chạy trên tokio multi-thread executor (đã lock từ B10 — tokio = { features = ["macros", "rt-multi-thread", "signal"] }). Executor sinh số worker thread tương đương số CPU logical: máy 8 core → 8 worker thread. Mỗi worker là một cooperative scheduler chạy task qua poll liên tục.

Mỗi handler là một async fn return Future — compiler tự sinh state machine implement trait Future. Worker pick handler, gọi poll; nếu Future Ready (xong), worker pick task khác; nếu Future Pending (chờ IO, DB query, network response), worker park task lại và pick task khác sẵn sàng.

Pattern này gọi là cooperative multitasking — task tự yield khi gặp await tại điểm I/O blocking, không phải kernel preempt như thread thông thường.

Timeline đơn giản cho hai request concurrent trên một worker thread:

Thread X | Request A      | Request B
---------+----------------+----------------
t=0      | start          |
t=1      | await DB query | (parking A)
t=2      |                | start
t=3      |                | await Redis
t=4      | DB ready,resume|
t=5      | response       | (parking B)
t=6      |                | Redis ready,resume
t=7      |                | response

Một worker phục vụ A và B đan xen — tổng thời gian wall-clock thấp hơn nhiều so với serial. Nếu handler là fn sync (không await), worker không thể yield giữa chừng → request B phải đợi A xong toàn bộ → throughput sụt thẳng.

So sánh với mô hình thread-per-request (Tomcat, Express cluster mode): mỗi request một thread OS riêng, context switch tốn kernel call, stack 1-8MB/thread → giới hạn ~1k thread/process. tokio cooperative scheduler chỉ tốn vài KB/task state machine, một worker phục vụ ~10k connection concurrent với latency thấp hơn.

Đây là lý do BẮT BUỘC handler axum là async fn. Compiler từ chối fn sync vì trait Handler yêu cầu return type implement Future (chi tiết bound ở step kế).

Series Rust Cơ Bản đã giới thiệu #[tokio::main] ở B251 (tokio runtime macro). Bài này đi sâu cách macro mở rộng trong context axum.

Code bạn viết với attribute macro:

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // body async của bạn
    app::run().await
}

Compiler desugar thành code synchronous đầy đủ:

fn main() -> anyhow::Result<()> {
    tokio::runtime::Builder::new_multi_thread()
        .enable_all()
        .build()
        .expect("build tokio runtime")
        .block_on(async {
            // body async của bạn
            app::run().await
        })
}

Phân tích từng phần:

• new_multi_thread() — chọn multi_thread flavor mặc định. Runtime sinh nhiều worker thread chia sẻ queue task qua work-stealing scheduler. Mặc định worker_threads bằng số CPU logical (num_cpus::get() dưới hood).
• enable_all() — bật mọi driver (IO driver cho socket, time driver cho sleep/interval, signal driver). Nếu chỉ cần IO không cần timer thì có enable_io() riêng, nhưng enable_all là default cho web server.
• build() — return Result<Runtime, io::Error>. Fail hiếm khi xảy ra (chỉ khi OS hết file descriptor cho IO driver).
• block_on(future) — chạy future cho tới khi Ready, block thread main trong lúc chờ. Đây là cầu nối sync world (main fn) và async world (handler).

Custom config khi cần override:

#[tokio::main(flavor = "multi_thread", worker_threads = 4)]
async fn main() -> anyhow::Result<()> {
    app::run().await
}

Flavor current_thread dùng cho test, embedded, CLI tool không cần concurrent — chỉ một thread duy nhất chạy executor, không cần Send bound (sẽ thấy ở step 4):

#[tokio::main(flavor = "current_thread")]
async fn main() -> anyhow::Result<()> { /* ... */ }

Shop API giữ default multi_thread với worker_threads tự động theo CPU — phù hợp production container cấp tài nguyên CPU động.

Handler axum follow pattern (đã lock từ B13):

async fn name(extractors...) -> impl IntoResponse

Return value sau khi compiler desugar async fn là một anonymous Future — state machine ẩn không có tên cụ thể. Future này phải implement hai trait bound bắt buộc:

• Send — type có thể move giữa thread an toàn. Cần thiết vì tokio multi_thread executor move task giữa worker thread qua work-stealing — task tạo ở worker 1 có thể tiếp tục poll ở worker 2 sau khi park-resume.
• 'static — không borrow data có lifetime ngắn hơn 'static. Executor lưu future trong queue task, không biết khi nào poll → không thể giữ reference tới stack frame ngoài.

Compiler enforce qua trait bound trong axum source (file axum/src/handler/mod.rs):

pub trait Handler<T, S>: Clone + Send + Sized + 'static {
    type Future: Future<Output = Response> + Send + 'static;
    fn call(self, req: Request, state: S) -> Self::Future;
}

Associated type Future bị bound Send + 'static. Compiler check khi đăng ký handler qua get(handler)/post(handler) — nếu handler return Future không Send, build error tại chính dòng .route(...).

Common error — capture reference qua await:

async fn bad_handler() -> String {
    let data = String::from("hello");
    let r: &String = &data;
    some_async_work().await;
    r.clone()  // !Send? compile error nếu r live qua await
}

Lỗi compiler chỉ vào dòng .route() đăng ký handler:

error: future cannot be sent between threads safely
    |
    | .route("/x", get(bad_handler))
    |              ^^^^^^^^^^^^^^^^ future returned by `bad_handler` is not `Send`

Fix bằng cách .to_owned() hoặc wrap Arc<T>:

async fn good_handler() -> String {
    let data = std::sync::Arc::new(String::from("hello"));
    let r = data.clone();
    some_async_work().await;
    (*r).clone()
}

Future của handler là Send nếu mọi data nó hold qua await đều là Send. Hai anti-pattern phổ biến và cách fix:

Anti-pattern 1 — hold Rc<T> qua await. Rc là reference counted non-thread-safe (counter không atomic) → không implement Send. Nếu hold qua await, Future trở thành !Send.

use std::rc::Rc;

async fn bad_rc() -> String {
    let data = Rc::new(String::from("hello"));
    tokio::time::sleep(std::time::Duration::from_millis(10)).await;
    // data vẫn live qua await ↑ → Future !Send
    (*data).clone()
}

Fix: dùng Arc<T> (atomic counter, implement Send + Sync):

use std::sync::Arc;

async fn good_arc() -> String {
    let data = Arc::new(String::from("hello"));
    tokio::time::sleep(std::time::Duration::from_millis(10)).await;
    (*data).clone()  // OK — Arc là Send
}

Anti-pattern 2 — hold MutexGuard từ std::sync::Mutex qua await. Vấn đề kép:

• MutexGuard không implement Send → Future !Send.
• Risk deadlock — nếu worker park task lúc đang hold guard, task khác chạy trên worker khác cố acquire mutex sẽ block. Nếu task khác đó chính là task cần để wake task hold guard → deadlock vĩnh viễn.

use std::sync::Mutex;

async fn bad_mutex(data: Arc<Mutex<Vec<u32>>>) -> usize {
    let guard = data.lock().unwrap();
    some_async_work().await;  // guard live qua await
    guard.len()
}

Hai cách fix:

Cách A — dùng tokio::sync::Mutex async-aware, lock().await trả guard implement Send:

use tokio::sync::Mutex;

async fn good_async_mutex(data: Arc<Mutex<Vec<u32>>>) -> usize {
    let guard = data.lock().await;
    some_async_work().await;
    guard.len()
}

Cách B — giải phóng guard trước await (recommended khi data không cần truy cập sau await):

async fn good_sync_mutex(data: Arc<Mutex<Vec<u32>>>) -> usize {
    let len = {
        let guard = data.lock().unwrap();
        guard.len()
    };  // guard drop ở đây
    some_async_work().await;
    len
}

Cách B thường hiệu năng cao hơn — std::sync::Mutex nhanh hơn tokio::sync::Mutex khi critical section ngắn. Cộng đồng Rust còn khuyến nghị parking_lot::Mutex cho sync mutex hiệu năng tốt hơn std mặc định.

Use case rất phổ biến trong API: handler tạo resource thành công và return response ngay, nhưng có vài tác vụ phụ không cần client chờ:

• Send email confirmation đơn hàng.
• Audit log ghi action vào table audit_logs.
• Cache invalidation các key liên quan.
• Webhook delivery retry cho partner.

Pattern fire-and-forget — spawn task background qua tokio::spawn, handler return ngay không await task xong:

use axum::Json;
use shop_common::error::AppResult;

async fn checkout(
    State(state): State<AppState>,
    Json(payload): Json<CheckoutDto>,
) -> AppResult<Json<Order>> {
    let order = create_order(&state.pool, payload).await?;

    // Fire-and-forget: spawn background task gửi email
    let order_id = order.id;
    tokio::spawn(async move {
        if let Err(e) = send_confirmation_email(order_id).await {
            tracing::error!(error = %e, order_id, "failed to send email");
        }
    });

    Ok(Json(order))  // Response trả ngay, không chờ email
}

Quan sát thiết kế:

• tokio::spawn(future) return JoinHandle<T> — đối tượng để await kết quả task. Fire-and-forget bỏ JoinHandle (drop ngay).
• JoinHandle drop không kill task — task vẫn chạy tới khi xong hoặc runtime shutdown. Khác biệt với std::thread::JoinHandle (drop không kill thread tương tự, nhưng không có cooperative shutdown).
• async move bắt buộc — move ownership data (order_id) vào task. Task có lifetime 'static, không borrow từ scope ngoài.
• Error trong task không bubble ra handler — phải log qua tracing::error! với context đầy đủ để observability bắt được.

Cảnh báo: spawn nhiều task không track gây khó shutdown gracefully. Khi K8s gửi SIGTERM, runtime stop tokio executor → task background đang chạy bị drop giữa chừng. Email có thể không gửi xong, audit log có thể không persist.

Production pattern Shop API:

• Task nhẹ, mất không đau (cache invalidation, metric increment) — dùng tokio::spawn đơn giản.
• Task quan trọng, cần guarantee delivery (email, webhook, payment notification) — dùng job queue (apalis + Redis backend, sẽ lock G21). Job persist vào Redis, worker riêng (shop-worker binary) pull job và retry exponential backoff khi fail.

Ban đầu Shop API có thể dùng tokio::spawn cho audit log và email confirmation (G20), G21 migrate sang queue khi domain phình lớn cần guarantee.

Hai hàm sleep tưởng tương tự nhưng hành vi đối lập:

• std::thread::sleep(Duration) — block toàn worker thread trong duration. Worker stop serving mọi task khác.
• tokio::time::sleep(Duration).await — yield task hiện tại, worker pick task khác. Sau khi timer ready, task được wake và resume.

Timeline so sánh blocking vs async cho hai request concurrent (mỗi cái sleep 100ms):

std::thread::sleep — worker blocked:
t=0       | A start
t=0-100ms | A blocking, B QUEUED, worker idle
t=100     | A return
t=100     | B start
t=100-200 | B blocking
t=200     | B return
Total: 200ms

tokio::time::sleep — worker yields:
t=0       | A start, A await sleep
t=0       | B start, B await sleep
t=0-100   | worker idle (cả A và B đang park)
t=100     | A wake, A return
t=100     | B wake, B return
Total: 100ms

CẤM dùng std::thread::sleep trong handler axum. Một request lỗi gọi std::thread::sleep(Duration::from_secs(5)) sẽ làm worker đó dừng phục vụ mọi request khác trong 5 giây.

Pattern blocking tương tự cần tránh — luôn dùng phiên bản async:

• std::fs::read_to_string(path) blocking → tokio::fs::read_to_string(path).await async.
• std::net::TcpStream::connect(...) blocking → tokio::net::TcpStream::connect(...).await async.
• reqwest::blocking::Client::get(...) blocking → reqwest::Client::get(...).send().await async.
• std::sync::mpsc::Receiver::recv() blocking → tokio::sync::mpsc::Receiver::recv().await async.

Quy tắc đơn giản: trong handler axum, nếu hàm có tên giống nhưng không .await được, nó đang blocking — tìm phiên bản async thay thế.

Có loại task không phải đợi IO nhưng vẫn blocking — CPU-bound. Code chiếm CPU 100ms thực tế cũng làm worker thread dừng poll task khác 100ms, dù không có std::thread::sleep.

Ví dụ task CPU-heavy phổ biến trong API:

• Hash password với argon2 (cố tình chậm để chống brute-force, ~100-500ms).
• Compress image, resize thumbnail.
• Render PDF invoice từ template.
• Parse JSON cực lớn (>10MB).
• Sign cryptography document.

Solution: tokio::task::spawn_blocking — chạy closure trên blocking thread pool riêng. Pool này có mặc định 512 thread (config qua max_blocking_threads), tách biệt hoàn toàn với worker async. Closure block thread pool blocking — worker async vẫn phục vụ request khác bình thường.

use axum::body::Bytes;
use shop_common::error::AppResult;

async fn upload_image(image: Bytes) -> AppResult<String> {
    // CPU-heavy: spawn lên blocking thread pool
    let processed = tokio::task::spawn_blocking(move || {
        // Block thread pool blocking — không ảnh hưởng worker async
        let img = image::load_from_memory(&image)?;
        let thumb = img.thumbnail(800, 800);
        let mut buf = Vec::new();
        thumb.write_to(&mut std::io::Cursor::new(&mut buf), image::ImageFormat::WebP)?;
        Ok::<_, anyhow::Error>(buf)
    })
    .await??;  // outer ? cho JoinError, inner ? cho anyhow::Error

    upload_to_s3(processed).await
}

Quan sát:

• Hai dấu ?? — outer cho JoinError (task panic hoặc cancel), inner cho lỗi business của closure.
• Closure phải Send + 'static — chuyển ownership data qua move tương tự tokio::spawn.
• Return value qua channel internal — tokio implement spawn_blocking qua oneshot channel: thread blocking write kết quả, async task await read kết quả.

Phân biệt rõ ba primitive thường nhầm lẫn:

• tokio::spawn(async {...}) — spawn async task lên worker async runtime. Closure phải là async block. Dùng cho fire-and-forget I/O bound task.
• tokio::task::spawn_blocking(|| {...}) — spawn sync closure lên blocking thread pool. Closure là sync function (không async). Dùng cho CPU-heavy hoặc gọi sync API legacy.
• std::thread::spawn(|| {...}) — spawn OS thread mới hoàn toàn ngoài tokio. Hiếm dùng trong axum handler, chỉ dùng cho background daemon dài hạn.

Shop API áp dụng spawn_blocking cho: hash password argon2 (B103), generate PDF invoice (nếu có), image processing (resize thumbnail upload product photo).

Thêm một demo route minh họa pattern fire-and-forget vào structure refactor sau B17. Route tạm thời để verify pattern end-to-end — sẽ remove khi G3+ có handler resource thật.

Tạo file mới crates/shop-api/src/routes/demo_async.rs:

// File: crates/shop-api/src/routes/demo_async.rs
use axum::{routing::post, Json, Router};
use serde_json::json;
use std::time::Duration;

use crate::state::AppState;

/// Build sub-router cho demo background task endpoint (B18).
///
/// TẠM THỜI — minh họa pattern fire-and-forget với `tokio::spawn`. Sẽ
/// REMOVE khi G3+ có handler resource thật. Lúc đó dòng
/// `.merge(routes::demo_async::routes())` trong `router.rs` cũng xóa
/// luôn cùng `pub mod demo_async;` trong `routes/mod.rs`.
pub fn routes() -> Router<AppState> {
    Router::new().route("/demo/background", post(demo_background))
}

async fn demo_background() -> Json<serde_json::Value> {
    // Fire-and-forget background task — handler return ngay không chờ.
    tokio::spawn(async move {
        tracing::info!("background task started");
        tokio::time::sleep(Duration::from_secs(5)).await;
        tracing::info!("background task completed");
    });

    Json(json!({
        "status": "accepted",
        "message": "task running in background"
    }))
}

Cập nhật crates/shop-api/src/routes/mod.rs thêm dòng aggregate module mới:

// File: crates/shop-api/src/routes/mod.rs

pub mod demo_async;
pub mod demo_error;
pub mod health;
pub mod version;

Cập nhật crates/shop-api/src/router.rs thêm dòng merge sub-router mới vào build_router:

// File: crates/shop-api/src/router.rs
pub fn build_router(state: AppState) -> Router {
    Router::new()
        .route("/", get(root))
        .merge(routes::health::routes())
        .merge(routes::version::routes())
        .merge(routes::demo_error::routes())
        .merge(routes::demo_async::routes())  // ← B18
        .with_state(state)
}

Build và test verify hành vi fire-and-forget:

cd shop
cargo run -p shop-api
# 2026-06-12T10:00:00Z  INFO shop_api::app: shop-api listening addr=0.0.0.0:3000

curl -X POST http://localhost:3000/demo/background
# {"message":"task running in background","status":"accepted"}
# (response trả ngay lập tức)

Log trên server hiển thị task chạy nền 5 giây sau khi handler đã return:

2026-06-12T10:00:05Z  INFO shop_api::routes::demo_async: background task started
(5 giây sau)
2026-06-12T10:00:10Z  INFO shop_api::routes::demo_async: background task completed

Pattern này lock cho Shop API: handler endpoint POST /checkout (G20) sẽ tạo order qua create_order().await?, sau đó tokio::spawn email confirmation, response 202 Accepted ngay. G21 migrate email sang Redis job queue (apalis) khi cần guarantee delivery + retry.

• axum BẮT BUỘC async fn handler — sync handler không support vì trait Handler yêu cầu return type implement Future.
• #[tokio::main] desugar thành Builder::new_multi_thread().enable_all().build()?.block_on(async { ... }) — multi_thread flavor mặc định, worker_threads tự động theo số CPU.
• Future trả về từ handler phải Send + 'static — compiler enforce qua trait Handler (đã preview B13). Future là Send nếu mọi data hold qua await đều Send.
• Send pitfall phổ biến: Rc<T> → fix bằng Arc<T>; std::sync::Mutex guard qua await → fix bằng tokio::sync::Mutex, hoặc giải phóng guard trước await.
• tokio::spawn(async move { ... }) cho fire-and-forget task nhẹ — handler return 202 Accepted ngay, task tiếp tục chạy nền. JoinHandle drop không kill task.
• tokio::time::sleep(Duration).await async yield worker; std::thread::sleep(Duration) blocking worker (CẤM dùng trong handler).
• CPU-heavy task → tokio::task::spawn_blocking(|| { ... }) chạy trên blocking thread pool riêng (mặc định 512 thread), không block worker async.
• Shop API: tokio::spawn cho audit log + email confirmation ban đầu (G20); G21 migrate sang Redis job queue (apalis) cho task quan trọng cần guarantee delivery + retry.
• Phân biệt 3 primitive: tokio::spawn (async task), tokio::task::spawn_blocking (CPU sync task), std::thread::spawn (OS thread ngoài tokio, hiếm dùng).
• Pattern blocking sync cần tránh trong handler: std::fs, std::net::TcpStream, reqwest::blocking, std::sync::mpsc::Receiver::recv(). Luôn tìm phiên bản async tương ứng (tokio::fs, tokio::net, reqwest::Client async, tokio::sync::mpsc).
• Demo route POST /demo/background minh họa fire-and-forget end-to-end trong file routes/demo_async.rs — sẽ remove khi G3+ có handler resource thật.

Tự trả lời, đáp án ở cuối:

1. Tại sao handler axum BẮT BUỘC async fn? Hậu quả nếu cố viết sync fn handler là gì? Compiler báo lỗi ở chỗ nào?
2. Trait bound Send + 'static trên Future của handler enforce ở đâu (file source axum)? Lỗi phổ biến gì khiến Future trở thành !Send và cách fix?
3. Phân biệt tokio::spawn vs tokio::task::spawn_blocking. Closure mỗi cái yêu cầu signature thế nào (async hay sync)? Khi nào dùng cái nào? Cho ví dụ task cụ thể Shop API.
4. std::thread::sleep(Duration::from_secs(5)) trong handler async axum gây vấn đề gì? Hậu quả production scale ra sao? Cách fix đúng?
5. Shop API endpoint POST /checkout tạo order xong muốn gửi email confirmation không chờ client. Code snippet pattern fire-and-forget với tokio::spawn trông như thế nào? Tại sao G21 lại migrate sang Redis job queue thay vì giữ tokio::spawn?

Bài 19: axum-extra: Cookies, Query, MultiPart, ... — giới thiệu axum-extra crate với extension features: cookie jar (PrivateCookieJar, SignedCookieJar), TypedHeader (Authorization, ContentType), Form extractor, erased-json — chuẩn bị cho extractor sâu ở Group 4.