Bài 38: Streaming Body — body::Body

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

• Hiểu khác biệt buffered body (Bytes, Json<T>) load full vào memory vs streaming body (Body) xử lý từng chunk khi đến.
• Biết extract streaming body qua axum::body::Body extractor + BodyExt::into_data_stream() consume từng chunk qua StreamExt::next().await.
• Áp dụng pattern stream upload large file với incremental size check (reject 413 ngay khi vượt, KHÔNG đợi load full).
• Hiểu Body::from_stream(stream) cho download response chunked transfer encoding tự động — wrap file qua tokio_util::io::ReaderStream.
• Nắm cơ chế backpressure qua bounded mpsc channel: server CHỈ đọc khi client ack chunk trước → memory cap predictable.
• Biết Server-Sent Events (SSE) qua axum::response::sse::Sse built-in cho real-time push event tới client.
• Pattern Shop API: admin CSV export 1M order rows streaming qua sqlx .fetch() B197, order updates SSE B189, product image upload B62.

Hai mô hình xử lý body khác nhau căn bản về memory profile và độ phức tạp code. Buffered (Bytes, String, Json<T>, Form<T>) load FULL body vào memory rồi parse 1 lần — đơn giản, error handling rõ ràng (parse fail trả 422 ngay), nhưng peak memory = body size → file 1GB nuốt 1GB RAM dễ OOM single-pod K8s. Streaming (Body) xử lý từng chunk khi đến — memory bounded ~1 chunk (8KB-1MB), latency thấp (xử lý song song với download tiếp tục), nhưng code phức tạp hơn và error giữa stream khó recover (đã ghi 500MB vào disk thì abort thế nào).

Buffered (Bytes/Json):
   [chunk1][chunk2][chunk3] → [FULL BODY trong RAM] → parse → handler
                              (peak memory = body size)

Streaming (Body):
   [chunk1] → handler.process(chunk1) → drop chunk1
   [chunk2] → handler.process(chunk2) → drop chunk2
   [chunk3] → handler.process(chunk3) → drop chunk3
              (peak memory = 1 chunk ~8KB-1MB)

Decision rule lock vĩnh viễn Shop API:

• File ≤ 5MB → buffered (Bytes/Json/AppMultipart B36) — code đơn giản parse 1 lần, error handling clean qua ? operator, peak memory predictable trong giới hạn cap. Use case: webhook payload Stripe ~10-100KB stay buffered (lock B37), product image admin upload ≤5MB (lock B36), JSON API endpoint baseline ≤2MB default.
• File > 5MB → streaming (Body) MANDATORY — memory bounded regardless file size, peak ~1MB chunk. Use case: admin CSV export 1M+ order rows ~200MB (B197), invoice PDF stream từ S3 future, log file download admin.
• Ngưỡng 5MB align DefaultBodyLimit::max(5 * 1024 * 1024) product image route layer lock B36 — vượt ngưỡng phải explicit chuyển sang streaming pattern hoặc tăng limit per route.

Streaming KHÔNG free lunch — trade-off với buffered: (a) error mid-stream khó recover (đã ghi 500MB vào S3 multipart upload rồi server lỗi → phải abort_multipart_upload cleanup tránh orphan part charge phí), (b) validate format khó (chỉ có chunk hiện tại, không biết toàn thể), (c) backpressure phải explicit handle (chi tiết bước 6). Decision rule trên cân bằng simplicity (buffered cho 90% endpoint) vs scalability (streaming cho 10% endpoint critical).

axum cung cấp axum::body::Body wrapper trait http_body::Body (từ http crate ecosystem tokio-rs) cho stream từng chunk. Pattern consume: convert qua body.into_data_stream() trả impl Stream<Item = Result<Bytes, axum::Error>> rồi loop iterate qua trait StreamExt::next().await:

// File: crates/shop-api/src/routes/admin/uploads.rs (B62 implement)
use axum::body::Body;
use axum::http::StatusCode;
use http_body_util::BodyExt;
use futures::stream::StreamExt;
use shop_common::error::{AppError, AppResult};

const MAX_UPLOAD_SIZE: u64 = 100 * 1024 * 1024; // 100MB

async fn upload_large_file(body: Body) -> AppResult<StatusCode> {
    let mut stream = body.into_data_stream();
    let mut total_bytes = 0u64;
    let mut file = tokio::fs::File::create("/tmp/upload.bin").await
        .map_err(|e| AppError::Internal(anyhow::anyhow!(e)))?;

    while let Some(chunk_result) = stream.next().await {
        let chunk = chunk_result
            .map_err(|e| AppError::BadRequest(format!("stream error: {}", e)))?;

        total_bytes += chunk.len() as u64;
        if total_bytes > MAX_UPLOAD_SIZE {
            return Err(AppError::BadRequest("file too large".to_string()));
        }

        tokio::io::AsyncWriteExt::write_all(&mut file, &chunk).await
            .map_err(|e| AppError::Internal(anyhow::anyhow!(e)))?;
    }

    Ok(StatusCode::CREATED)
}

Điểm chú ý:

• body: Body đặt CUỐI arg list theo lock B31 (cùng quy tắc Json/Form/Multipart/Bytes B32/B35/B36/B37) — Body consume request body one-shot qua FromRequest.
• body.into_data_stream() đến từ trait http_body_util::BodyExt — workspace dep sẽ add http-body-util = "0.1" khi B62 implement.
• StreamExt::next() đến từ futures::stream::StreamExt trait — workspace dep futures = "0.3"; alternative tokio_stream::StreamExt compat tương đương.
• Mỗi chunk là bytes::Bytes cheap-clone Arc internal (đã giới thiệu B37) — chunk size phụ thuộc hyper transport layer (HTTP/1.1 chunked encoding ~8-32KB typical, HTTP/2 frame ~16KB) KHÔNG control trực tiếp từ application.
• Incremental size check qua total_bytes > MAX_UPLOAD_SIZE reject ngay khi vượt — KHÔNG đợi load hết body như Bytes extractor (tiết kiệm bandwidth + memory cho attack DoS spam request lớn).
• DefaultBodyLimit::max(N) route layer vẫn áp dụng cho Body extractor — server abort connection khi vượt limit trước cả khi handler thấy chunk.

Pitfall lớn nhất: handler extract Bytes cho file 1GB — axum load full Body vào memory rồi materialize thành bytes::Bytes → process bị OOM kill bởi K8s (memory limit container thường 512MB-2GB). Pitfall xảy ra khi developer copy-paste pattern Bytes từ webhook B37 cho mọi endpoint upload mà không kiểm tra size budget.

Fix: extract Body → stream chunk → incremental size check (đã thấy ở bước 3). Save destination 2 lựa chọn:

(a) Save to disk qua tokio::fs::File + AsyncWriteExt::write_all — phù hợp dev/single-pod tạm; production tránh vì disk local pod restart mất file (lock B36 disk local CHỈ dev).

(b) Upload S3 multipart streaming — pattern production: chia file thành parts ≥5MB (S3 minimum part size hard limit) upload đồng thời, complete khi xong:

// File: crates/shop-api/src/routes/admin/uploads.rs (B62 implement)
// Pseudo-code S3 multipart streaming — thực tế B62 dùng object_store crate
// abstract chi tiết (lock B36 lean về object_store cho R2 swap)

const PART_SIZE: usize = 5 * 1024 * 1024; // 5MB S3 minimum

async fn upload_to_s3_streaming(
    body: Body,
    s3: &S3Client,
    bucket: &str,
    key: &str,
) -> AppResult<()> {
    let upload_id = s3.create_multipart_upload(bucket, key).await?;
    let mut stream = body.into_data_stream();
    let mut buffer: Vec<u8> = Vec::with_capacity(PART_SIZE);
    let mut part_number = 1i32;
    let mut completed_parts = Vec::new();

    while let Some(chunk_result) = stream.next().await {
        let chunk = chunk_result?;
        buffer.extend_from_slice(&chunk);

        // Upload mỗi 5MB (S3 minimum part size, trừ part cuối)
        if buffer.len() >= PART_SIZE {
            let etag = s3.upload_part(
                bucket,
                key,
                &upload_id,
                part_number,
                buffer.clone().into(),
            ).await?;
            completed_parts.push((part_number, etag));
            buffer.clear();
            part_number += 1;
        }
    }

    // Upload buffer còn lại (part cuối có thể nhỏ hơn 5MB)
    if !buffer.is_empty() {
        let etag = s3.upload_part(bucket, key, &upload_id, part_number, buffer.into()).await?;
        completed_parts.push((part_number, etag));
    }

    s3.complete_multipart_upload(bucket, key, &upload_id, completed_parts).await?;
    Ok(())
}

Shop API B62 implement product image upload qua object_store crate đã handle multipart S3 abstraction tự động (lock B36) — handler chỉ stream chunk vào BufWriter wrap object_store::PutMultipartWriter, object_store quản phần boundary 5MB internal. Tránh tự implement multipart trừ khi cần fine-grained control (vd parallel upload nhiều part đồng thời tăng throughput).

Error mid-stream lưu ý cleanup orphan part qua s3.abort_multipart_upload(upload_id) nếu lỗi sau khi đã upload vài part — S3 charge phí storage cho orphan part đến khi cleanup (default lifecycle policy 7 ngày).

Streaming KHÔNG chỉ cho upload — response cũng có thể stream chunk thay vì 1 response body lớn. Use case: large file download (video, log, PDF invoice 100MB+), Server-Sent Events real-time push. Pattern: tạo Stream<Item = Result<Bytes, E>> rồi wrap qua Body::from_stream(stream):

// File: crates/shop-api/src/routes/admin/logs.rs (future implement)
use axum::body::Body;
use axum::http::header;
use axum::response::IntoResponse;
use tokio_util::io::ReaderStream;
use shop_common::error::{AppError, AppResult};

async fn download_log() -> AppResult<impl IntoResponse> {
    let log_path = "/var/log/shop-api.log";
    let file = tokio::fs::File::open(log_path).await
        .map_err(|_| AppError::NotFound)?;

    // ReaderStream: AsyncRead → Stream<Item = Result<Bytes, io::Error>>
    let stream = ReaderStream::new(file);
    let body = Body::from_stream(stream);

    Ok((
        [
            (header::CONTENT_TYPE, "text/plain"),
            (header::CONTENT_DISPOSITION, "attachment; filename=\"shop-api.log\""),
        ],
        body,
    ))
}

Điểm chú ý:

• tokio_util::io::ReaderStream cầu nối AsyncRead (trait base của tokio::fs::File, tokio::net::TcpStream, S3 GetObject response stream) sang Stream<Item = Result<Bytes, std::io::Error>> — workspace dep tokio-util = { version = "0.7", features = ["io"] }.
• Chunk size default ReaderStream 8KB; cấu hình qua ReaderStream::with_capacity(file, 64 * 1024) nếu cần balance throughput vs memory.
• Body::from_stream(stream) wrap Stream thành axum::body::Body implement IntoResponse; axum + hyper tự set Transfer-Encoding: chunked response header — KHÔNG cần Content-Length (server không biết trước file size).
• Browser/curl nhận chunked transfer encoding incremental — TTFB (Time to First Byte) chỉ ~50ms (open file + send headers), KHÔNG phải đợi server đọc xong full file mới gửi response.
• Content-Disposition: attachment; filename="..." trigger browser download dialog thay vì inline render (security: chống XSS HTML upload tương tự lock B36).

Use case Shop API future: invoice PDF download stream từ S3 — s3.get_object(bucket, key) trả ByteStream implement Stream<Bytes>, wrap thẳng qua Body::from_stream proxy bytes từ S3 sang client KHÔNG buffer toàn file vào RAM server.

Pitfall: server đọc file 10GB nhanh từ disk/S3 (1GB/s SSD NVMe) nhưng client download chậm qua mobile network (1MB/s) → nếu server tự ý đọc trước hết file vào buffer chờ client → server buffer hết 10GB RAM → OOM. Vấn đề kinh điển fast producer + slow consumer.

Backpressure = cơ chế producer (server đọc file) tự dừng đọc khi consumer (client download) chưa kịp xử lý chunk trước. axum + hyper tự handle qua tower::Service design pull-based (consumer pull data thay vì producer push) — khi consumer chưa poll_next Stream thì producer task tự suspend tại tx.send().await. Nhưng để có fine-grained control + memory cap predictable, dùng explicit bounded channel tokio mpsc:

use tokio::sync::mpsc;
use tokio_stream::wrappers::ReceiverStream;
use axum::body::Body;
use bytes::Bytes;

async fn stream_large_file_with_backpressure() -> impl IntoResponse {
    // Bounded channel: capacity 10 chunk → memory cap = 10 × chunk_size
    let (tx, rx) = mpsc::channel::<Result<Bytes, std::io::Error>>(10);

    // Spawn producer task đọc file vào channel
    tokio::spawn(async move {
        let mut file = tokio::fs::File::open("/data/big.bin").await.unwrap();
        let mut buf = vec![0u8; 1024 * 1024]; // 1MB chunk

        loop {
            let n = match tokio::io::AsyncReadExt::read(&mut file, &mut buf).await {
                Ok(0) => break, // EOF
                Ok(n) => n,
                Err(e) => { let _ = tx.send(Err(e)).await; return; }
            };
            // send().await tự block khi rx chưa consume (channel full)
            // → backpressure tự nhiên: producer dừng khi consumer chậm
            if tx.send(Ok(Bytes::copy_from_slice(&buf[..n]))).await.is_err() {
                return; // receiver dropped (client disconnect)
            }
        }
    });

    // ReceiverStream: mpsc::Receiver → Stream
    let stream = ReceiverStream::new(rx);
    Body::from_stream(stream)
}

Phân tích:

• Bounded channel capacity 10 — producer task block tại tx.send().await khi đã có 10 chunk pending trong channel, đợi consumer (client) lấy chunk ra qua poll Stream mới send tiếp.
• Memory cap predictable = capacity × chunk_size = 10 × 1MB = 10MB peak per request regardless file size (10GB hay 100MB đều cap 10MB).
• tokio_stream::wrappers::ReceiverStream wrap mpsc::Receiver implement Stream — axum poll Stream qua hyper response writer, client TCP slow → hyper slow poll → ReceiverStream slow poll → channel chậm drain → producer block — backpressure propagate end-to-end tự động.
• Client disconnect → axum drop response body → hyper drop ReceiverStream → rx drop → tx.send() trả Err → producer task return cleanup (giải phóng file handle + memory ngay).
• Capacity 10 là default Shop API lock — balance throughput (lớn hơn 1 cho phép producer làm việc trước consumer) vs memory cap (nhỏ hơn 100 tránh peak quá lớn). Tăng capacity nếu cần throughput cao với file nhỏ, giảm nếu memory critical.

Backpressure quan trọng cho large file stream + SSE long-lived connection — nếu KHÔNG bound, attacker mở 100 connection download chậm sẽ ép server allocate 100 × file_size RAM → DoS.

SSE (Server-Sent Events, chuẩn HTML5 W3C 2015) là pattern server push events tới client liên tục qua HTTP streaming long-lived connection. Khác WebSocket: SSE 1 chiều server → client (đủ cho notification/update real-time), reuse HTTP/1.1 + HTTP/2 native (proxy/CDN/firewall friendly), browser tự reconnect khi disconnect, format text-based plain dễ debug.

Content-Type response: text/event-stream. Format mỗi event: data: <json>\n\n (2 newline kết thúc event). axum có module built-in axum::response::sse::Sse wrap Stream của Event trả response chuẩn:

// File: crates/shop-api/src/routes/admin/notifications.rs (B189 implement)
use axum::extract::{Path, State};
use axum::response::sse::{Event, KeepAlive, Sse};
use futures::stream::Stream;
use std::convert::Infallible;
use std::time::Duration;
use crate::AppState;

async fn order_updates(
    State(state): State<AppState>,
    Path(order_id): Path<i64>,
) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
    // Subscribe Redis pub/sub channel "order:status:{id}" (B189 implement)
    let mut rx = state.redis.subscribe(&format!("order:status:{}", order_id)).await;

    let stream = async_stream::stream! {
        while let Some(payload) = rx.recv().await {
            let event = Event::default().data(payload);
            yield Ok(event);
        }
    };

    Sse::new(stream).keep_alive(
        KeepAlive::new()
            .interval(Duration::from_secs(30))
            .text("keep-alive")
    )
}

Điểm chú ý:

• Sse::new(stream) trả response Content-Type: text/event-stream + Cache-Control: no-cache tự động — KHÔNG cần build header manual.
• KeepAlive::interval(Duration::from_secs(30)) gửi comment line : keep-alive\n\n mỗi 30s để chống proxy NGINX/CloudFlare/HAProxy idle timeout default 60s đóng connection — MANDATORY cho SSE long-lived connection production.
• Item type Result<Event, Infallible> vì stream KHÔNG fail (Redis subscribe loop forever); nếu Redis error retry transparent qua client browser tự reconnect mà.
• Client browser dùng new EventSource("/api/v1/notifications/order/123") nhận event qua onmessage handler — auto reconnect khi connection drop với last-event-id resume.
• Backpressure SSE: nếu Redis pub publish nhanh hơn client consume → bounded channel mpsc capacity nhỏ giữa Redis subscriber và Sse stream (pattern bước 6) tránh OOM.

Shop API use case lock: real-time order status update B189 — customer/admin xem order list page với SSE stream từ Redis pub/sub channel khi order chuyển trạng thái (pending → paid → shipping → delivered), notification toast hiện ngay không cần F5 reload. SSE pattern + keepalive deep dive chi tiết B50 module riêng.

Use case kinh điển streaming response Shop API: admin export 1M+ order rows ra CSV. Pattern buffered (build String 200MB rồi gửi) → OOM single-pod K8s + TTFB 30s+ (admin nhìn loading spinner mãi mới thấy download bắt đầu). Pattern streaming MANDATORY:

// File: crates/shop-api/src/routes/admin/orders.rs (B197 implement)
use axum::body::Body;
use axum::extract::State;
use axum::http::header;
use axum::response::IntoResponse;
use bytes::Bytes;
use futures::stream::StreamExt;
use shop_common::error::AppError;
use crate::extractors::RequireRole; // B135 RBAC
use crate::AppState;

async fn export_orders_csv(
    State(state): State<AppState>,
    _admin: RequireRole<"admin">,
) -> impl IntoResponse {
    // sqlx .fetch() trả Stream lazy load từng row từ Postgres cursor
    let row_stream = sqlx::query!(
        "SELECT id, customer_id, total, status, created_at \
         FROM orders ORDER BY created_at DESC"
    )
    .fetch(&state.pool);

    // Stream header CSV trước, sau đó stream từng row
    let header_row = "id,customer_id,total,status,created_at\n".to_string();
    let header_stream = futures::stream::once(async move {
        Ok::<_, AppError>(Bytes::from(header_row))
    });

    let body_stream = row_stream.map(|row_result| {
        let row = row_result
            .map_err(|e| AppError::Internal(anyhow::anyhow!(e)))?;
        let csv_row = format!(
            "{},{},{},{},{}\n",
            row.id, row.customer_id, row.total, row.status,
            row.created_at.to_rfc3339(),
        );
        Ok::<_, AppError>(Bytes::from(csv_row))
    });

    let full_stream = header_stream.chain(body_stream);

    let filename = format!("orders-{}.csv", chrono::Utc::now().format("%Y%m%d"));
    (
        [
            (header::CONTENT_TYPE, "text/csv; charset=utf-8"),
            (
                header::CONTENT_DISPOSITION,
                &format!("attachment; filename=\"{}\"", filename),
            ),
        ],
        Body::from_stream(full_stream),
    )
}

Phân tích:

• sqlx::query!().fetch(&pool) trả impl Stream<Item = Result<Row, sqlx::Error>> — lazy load từng row qua Postgres cursor transaction, KHÔNG materialize Vec 1M row vào RAM. So với .fetch_all() trả Vec<Row> load full → OOM trên 1M row.
• Memory bounded ~100KB peak (vài chục row chunk cộng channel buffer) regardless 1M hay 10M row.
• TTFB ~50ms — server gửi header response + CSV header row ngay sau query setup, KHÔNG đợi load full DB.
• Client browser tải về incremental, hiện progress bar bytes downloaded, save file disk từng phần — admin thấy file đang download ngay.
• RequireRole<"admin"> extractor RBAC B135 enforce quyền admin trước khi run query expensive.
• Production lưu ý: query DB hold connection lâu (1M row × 1ms/row = 1000s = ~17 phút) → set statement_timeout Postgres lớn hơn, dùng read replica tránh ảnh hưởng pool primary, monitor pool starvation Prometheus.

Pattern lock vĩnh viễn Shop API: mọi export admin lớn (orders, customers, payments, inventory transactions) MANDATORY streaming qua sqlx .fetch() + Body::from_stream, KHÔNG buffer toàn data. Crate csv = "1" alternative format chuẩn handle escape comma/quote/newline đúng — B197 chọn cuối khi implement.

• Buffered (Bytes, Json<T>, Form<T>): load FULL body vào RAM → simple, OK cho file ≤5MB (webhook, JSON API endpoint, product image lock B36).
• Streaming (Body): xử lý từng chunk khi đến → memory bounded ~1 chunk (8KB-1MB), OK cho file >5MB; trade-off code phức tạp + error mid-stream khó recover.
• Decision rule lock vĩnh viễn: ≤5MB → buffered, >5MB → streaming MANDATORY; ngưỡng align body size limit product image B36.
• axum::body::Body extractor + BodyExt::into_data_stream() + loop StreamExt::next().await consume từng chunk Bytes cheap-clone.
• Upload streaming pattern 4 bước: extract Body CUỐI arg list → into_data_stream → loop chunk + incremental size check (reject 413 ngay khi vượt) → save disk hoặc S3 multipart 5MB minimum part size qua object_store (lock B36).
• Download streaming pattern: Body::from_stream(stream) wrap Stream → axum + hyper set Transfer-Encoding: chunked tự động; cầu nối từ file/TcpStream/S3 qua tokio_util::io::ReaderStream::new(reader).
• Backpressure: bounded mpsc channel tokio::sync::mpsc::channel(10) → producer block khi consumer chậm → memory cap predictable = capacity × chunk_size (vd 10 × 1MB = 10MB peak per request).
• SSE: axum::response::sse::Sse built-in wrap Stream của Event + KeepAlive::interval(30s) chống proxy idle timeout — B50 chi tiết.
• CSV export pattern: sqlx .fetch() lazy stream row từ Postgres cursor → map format CSV → Body::from_stream → peak memory ~100KB regardless 1M row, TTFB ~50ms.
• Pattern Shop API lock: admin CSV export B197 (streaming MANDATORY), order updates SSE B189 (Redis pub/sub forward), product image upload streaming B62 (object_store multipart S3).
• Workspace.dependencies preview B62/B189/B197 add: http-body-util = "0.1" (BodyExt::into_data_stream), futures = "0.3" (StreamExt::next), tokio-util = { version = "0.7", features = ["io"] } (ReaderStream cầu nối AsyncRead ↔ Stream).
• B38 conceptual + preview pattern, KHÔNG tạo file thực tế ở Shop API (Workspace State KHÔNG đổi). File implementation + workspace dep sẽ add ở B62 (product image), B189 (SSE order), B197 (CSV export).

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

1. Phân biệt buffered (Bytes) vs streaming (Body). Peak memory mỗi loại bằng gì? Decision rule Shop API khi nào dùng cái nào và tại sao ngưỡng 5MB?
2. Upload file 1GB với Bytes extractor → vấn đề gì? Fix bằng pattern nào? Liệt kê 4 bước pattern fix MANDATORY.
3. Body::from_stream(stream) cần Stream type với Item là gì? Header response axum + hyper tự set là gì? Cầu nối tokio::fs::File → Stream qua type nào?
4. Backpressure là gì? Tại sao fast producer + slow consumer gây OOM? Bounded mpsc channel giải quyết thế nào? Memory cap predictable bằng công thức nào?
5. CSV export 1M order rows. Pattern streaming so với load full vào RAM khác nhau gì về peak memory + TTFB + risk pod K8s? sqlx .fetch() vs .fetch_all() khác gì?

Bài 39: Extension Extractor — Request-Scoped Data — chi tiết Extension<T> set qua middleware (X-Request-Id middleware sẽ implement đầy đủ với tower-http::SetRequestIdLayer + custom layer wrap), extract trong handler qua Extension(req_id): Extension<RequestId>; so sánh với State<T> (Extension dynamic per-request lifetime, State persistent global Arc cloned cross-handler); use case: request_id propagation, trace context OpenTelemetry, current_user injection sau auth middleware, A/B test variant context.