Rust in Everything
基于 tower-http 0.7.1

其余常用中间件

主线位置：请求主线上的几层中间件，都用 Router::layer 套上。每一节只回答三个问题：它在请求经过的哪个点上起作用（call 时、响应 future 完成时、还是响应 body 流动时），它自己会返回什么，它看不到什么。

flowchart TB
  req["请求进来"] --> cors{"CORS：OPTIONS"}
  cors -- 是 --> pre["直接回预检响应"]
  cors -- 否 --> lim{"Limit：长度超限"}
  lim -- 是 --> r413["413"]
  lim -- 否 --> wrap["body 套上 Limited"]
  wrap --> tmo["Timeout：计时开始"]
  tmo --> cp["CatchPanic：捕获 panic"]
  cp --> handler["路由、提取器、handler"]
  handler --> resp["响应头出来"]
  resp --> trace["Trace：on_response"]
  trace --> bodyflow["响应 body 流动"]
  bodyflow --> eos["Trace：on_body_chunk、on_eos"]

图里的先后只是为了画出每层起作用的位置，实际顺序取决于你 .layer 的顺序：Router::layer 后套的在外层（见 layer 与 route_layer）。

示例用这一组依赖：

axum = "=0.8.9"
tower-http = { version = "=0.7.1", features = ["timeout", "limit", "cors", "trace", "catch-panic"] }
tower = { version = "=0.5.3", features = ["util"] }
http-body-util = "0.1"
futures = "0.3"
tracing = "0.1"
tokio = { version = "=1.54.1", features = ["macros", "rt-multi-thread", "test-util"] }

所有输出都在 M2 Max、rustc 1.99.0、--release 下跑 3 次，结果相同。用到暂停时钟（start_paused = true）的地方，毫秒数是 tokio 的虚拟时间，不受机器快慢影响。

TimeoutLayer：超时变成一个响应

tower 自己的 Timeout 超时返回 Err(Elapsed)，错误类型变成 BoxError，在 axum 里要配 HandleErrorLayer（layer 与 route_layer 最后一节）。tower-http 的 Timeout 不改错误类型，超时时造一个响应：

impl<S, ReqBody, ResBody> Service<Request<ReqBody>> for Timeout<S>
where
    S: Service<Request<ReqBody>, Response = Response<ResBody>>,
    ResBody: Default,
{
    type Response = S::Response;
    type Error = S::Error;
    type Future = ResponseFuture<S::Future>;
    // ...
    fn call(&mut self, req: Request<ReqBody>) -> Self::Future {
        let sleep = tokio::time::sleep(self.timeout);
        ResponseFuture {
            inner: self.inner.call(req),
            sleep,
            status_code: self.status_code,
        }
    }
}

ResponseFuture::poll：

        if this.sleep.poll(cx).is_ready() {
            let mut res = Response::new(B::default());
            *res.status_mut() = *this.status_code;
            return Poll::Ready(Ok(res));
        }

        this.inner.poll(cx)
  • 响应 body 用 B::default() 造，所以要求 ResBody: Default；axum 的 Body 满足，结果是空 body、没有任何额外的头。
  • 状态码由 TimeoutLayer::with_status_code 指定。旧的 TimeoutLayer::new 固定用 408，0.6.7 起已标为 deprecated（见下面的冷知识）。
  • 计时从 call 开始。这时 hyper 已经读完了请求头，所以读请求头的时间不算在内；那一段由 hyper 的 header_read_timeout 管（而它在 axum::serve 下默认不生效，见 hyper 卷 header_read_timeout 不生效）。
  • 先 poll sleep 再 poll 内层，和 tower 的 Timeout 相反（tower 卷 Timeout 先内层后 Sleep）。两边同时就绪时，这里是超时赢。
  • 超时之后内层 future 随 ResponseFuture 一起被 drop，handler 停在它最后一个 .await 处，不会继续执行。

「超时包不包括读 body」要分开看：请求 body 是 handler（或提取器）在内层 future 里读的，算；响应 body 是在 future 完成、响应头交出去之后才流动的，不算。

use axum::{
  Router,
  body::{Body, Bytes},
  http::{Request, StatusCode},
  routing::{get, post},
};
use http_body_util::BodyExt;
use std::time::Duration;
use tokio::time::{Instant, sleep};
use tower::ServiceExt;
use tower_http::timeout::TimeoutLayer;

// a stream that yields `n` chunks, one every `gap` ms
fn trickle(n: u64, gap: u64) -> Body {
  Body::from_stream(futures::stream::unfold(0, move |i| async move {
    if i == n {
      return None;
    }
    sleep(Duration::from_millis(gap)).await;
    Some((Ok::<_, std::io::Error>(Bytes::from_static(b"x")), i + 1))
  }))
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
  let app = Router::new()
    .route("/slow", get(|| async {
      sleep(Duration::from_millis(150)).await;
      "done"
    }))
    .route("/upload", post(|body: Bytes| async move { format!("got {}", body.len()) }))
    .route("/download", get(|| async { trickle(5, 100) }))
    .layer(TimeoutLayer::with_status_code(StatusCode::REQUEST_TIMEOUT, Duration::from_millis(100)));

  let cases = [
    ("GET /slow", Request::get("/slow").body(Body::empty()).unwrap()),
    ("POST /upload, body 3 x 20 ms", Request::post("/upload").body(trickle(3, 20)).unwrap()),
    ("POST /upload, body 3 x 50 ms", Request::post("/upload").body(trickle(3, 50)).unwrap()),
    ("GET /download, 5 x 100 ms", Request::get("/download").body(Body::empty()).unwrap()),
  ];
  for (name, req) in cases {
    let start = Instant::now();
    let res = app.clone().oneshot(req).await.unwrap();
    let head_at = start.elapsed().as_millis();
    let status = res.status();
    let body = res.into_body().collect().await.unwrap().to_bytes();
    println!(
      "{name:<30} {status} head at {head_at} ms, body {:?} done at {} ms",
      String::from_utf8_lossy(&body),
      start.elapsed().as_millis()
    );
  }
}

GET /slow                      408 Request Timeout head at 100 ms, body "" done at 100 ms
POST /upload, body 3 x 20 ms   200 OK head at 60 ms, body "got 3" done at 60 ms
POST /upload, body 3 x 50 ms   408 Request Timeout head at 100 ms, body "" done at 100 ms
GET /download, 5 x 100 ms      200 OK head at 0 ms, body "xxxxx" done at 500 ms
  • 上传 body 用 150 毫秒才送完，Bytes 提取器还在读，100 毫秒时整个请求变成 408。
  • 下载的响应头 0 毫秒就出去了，body 又流了 500 毫秒，TimeoutLayer 管不到。要限制 body 的传输时间，用同一模块里的 RequestBodyTimeoutLayer / ResponseBodyTimeoutLayer（每来一帧重新计时，抓「停住不动」）或 0.7.0 新加的 RequestBodyDeadlineLayer / ResponseBodyDeadlineLayer（#688，从头到尾只计一次时，抓「一字节一字节地挤」）。它们包的是 body 而不是 future，超时表现为 body 返回错误，而不是一个新的响应。
📝NOTE
冷知识：408 曾经是写死的

0.6.7 之前 TimeoutLayer 只有 new(timeout)，超时一律回 408 Request Timeout。issue #300「Should TimeoutLayer middleware allow returning other status codes?」指出 408 的语义不对：「a 408, which is currently returned, is meant to communicate: that the server did not receive a complete request message within the time that it was prepared to wait」，而超时多半是服务端处理慢。有人提过直接改成 503（#579，没有合并），最后的做法是 #599「Add support for custom status code in TimeoutLayer」：加 with_status_code，把 new 标为 deprecated（PR 说明：「Implement TimeoutLayer::with_status_code」「Mark TimeoutLayer::new as deprecated」）。默认值没改，只是不再替你选。随 0.6.7 发布，CHANGELOG：「TimeoutLayer::new() should be replaced with TimeoutLayer::with_status_code(). (Previously was StatusCode::REQUEST_TIMEOUT)」。所以源码里 new 还在，就是 with_status_code(StatusCode::REQUEST_TIMEOUT, timeout)，用它会得到一个编译警告。

RequestBodyLimitLayer 与 DefaultBodyLimit

axum 自己的上限 DefaultBodyLimit 只是往请求里放一个扩展，等 Bytes、Json 这类提取器读 body 时才套上 Limited，直接拿 Request 自己读 body 的不受限（body 提取器）。RequestBodyLimit 在中间件层动手，它的 call：

    fn call(&mut self, req: Request<ReqBody>) -> Self::Future {
        let content_length = req
            .headers()
            .get(http::header::CONTENT_LENGTH)
            .and_then(|value| value.to_str().ok()?.parse::<usize>().ok());

        let body_limit = match content_length {
            Some(len) if len > self.limit => return ResponseFuture::payload_too_large(),
            Some(len) => self.limit.min(len),
            None => self.limit,
        };

        let req = req.map(|body| Limited::new(http_body_util::Limited::new(body, body_limit)));

        ResponseFuture::new(self.inner.call(req))
    }
  • Content-Length 超过上限：内层根本不调用，直接回 413，正文是 create_error_response 里固定的 length limit exceeded，Content-Type: text/plain; charset=utf-8。body 一个字节都没读。
  • 否则给 body 套上 http-body-util 的 Limited，额度取上限和 Content-Length 中较小的那个。之后谁读 body，谁就会在超额时拿到 LengthLimitError。这一层自己不读 body，也不会因为流式 body 超额而回 413，回什么取决于读 body 的那一方。

于是在 axum 里两层叠在一起时：

use axum::{
  Router,
  body::{Body, Bytes},
  extract::{DefaultBodyLimit, Request},
  routing::post,
};
use http_body_util::BodyExt;
use std::sync::atomic::{AtomicUsize, Ordering};
use tower::ServiceExt;
use tower_http::limit::RequestBodyLimitLayer;

static CALLS: AtomicUsize = AtomicUsize::new(0);

async fn bytes(b: Bytes) -> String {
  CALLS.fetch_add(1, Ordering::SeqCst);
  format!("got {}", b.len())
}

async fn raw(req: Request) -> String {
  CALLS.fetch_add(1, Ordering::SeqCst);
  match req.into_body().collect().await {
    Ok(c) => format!("got {}", c.to_bytes().len()),
    Err(e) => format!("handler saw error: {e}"),
  }
}

// a body without size hint, so neither layer can know its length in advance
fn stream_body(len: usize) -> Body {
  let chunks: Vec<Result<Bytes, std::io::Error>> =
    vec![b'a'; len].chunks(1024).map(|c| Ok(Bytes::copy_from_slice(c))).collect();
  Body::from_stream(futures::stream::iter(chunks))
}

async fn send(app: &Router, name: &str, len: usize, content_length: Option<usize>) {
  CALLS.store(0, Ordering::SeqCst);
  let mut req = Request::post("/");
  if let Some(cl) = content_length {
    req = req.header("content-length", cl);
  }
  let res = app.clone().oneshot(req.body(stream_body(len)).unwrap()).await.unwrap();
  let status = res.status();
  let ct = res.headers().get("content-type").map(|v| v.to_str().unwrap().to_owned()).unwrap_or("-".into());
  let body = res.into_body().collect().await.unwrap().to_bytes();
  println!(
    "{name:<34} {status} handler={} type={ct} body={:?}",
    CALLS.load(Ordering::SeqCst),
    String::from_utf8_lossy(&body)
  );
}

#[tokio::main]
async fn main() {
  const KIB: usize = 1024;
  const MIB: usize = 1024 * 1024;
  let small = Router::new().route("/", post(bytes)).layer(RequestBodyLimitLayer::new(4 * KIB));
  send(&small, "4 KiB layer, 8 KiB, Content-Length", 8 * KIB, Some(8 * KIB)).await;
  send(&small, "4 KiB layer, 8 KiB, no length", 8 * KIB, None).await;
  send(&small, "4 KiB layer, 8 KiB, CL says 100", 8 * KIB, Some(100)).await;

  let raw_app = Router::new().route("/", post(raw)).layer(RequestBodyLimitLayer::new(4 * KIB));
  send(&raw_app, "4 KiB layer, Request handler", 8 * KIB, None).await;

  let big = Router::new().route("/", post(bytes)).layer(RequestBodyLimitLayer::new(4 * MIB));
  send(&big, "4 MiB layer, 3 MiB, no length", 3 * MIB, None).await;
  let big_off = Router::new()
    .route("/", post(bytes))
    .layer(DefaultBodyLimit::disable())
    .layer(RequestBodyLimitLayer::new(4 * MIB));
  send(&big_off, "4 MiB layer + disable(), 3 MiB", 3 * MIB, None).await;
  send(&big_off, "4 MiB layer + disable(), 5 MiB", 5 * MIB, None).await;
}

4 KiB layer, 8 KiB, Content-Length 413 Payload Too Large handler=0 type=text/plain; charset=utf-8 body="length limit exceeded"
4 KiB layer, 8 KiB, no length      413 Payload Too Large handler=0 type=text/plain; charset=utf-8 body="Failed to buffer the request body: length limit exceeded"
4 KiB layer, 8 KiB, CL says 100    413 Payload Too Large handler=0 type=text/plain; charset=utf-8 body="Failed to buffer the request body: length limit exceeded"
4 KiB layer, Request handler       200 OK handler=1 type=text/plain; charset=utf-8 body="handler saw error: length limit exceeded"
4 MiB layer, 3 MiB, no length      413 Payload Too Large handler=0 type=text/plain; charset=utf-8 body="Failed to buffer the request body: length limit exceeded"
4 MiB layer + disable(), 3 MiB     200 OK handler=1 type=text/plain; charset=utf-8 body="got 3145728"
4 MiB layer + disable(), 5 MiB     413 Payload Too Large handler=0 type=text/plain; charset=utf-8 body="Failed to buffer the request body: length limit exceeded"

谁先回答、回什么：

情况谁回答正文
有 Content-Length 且超过 tower-http 的上限RequestBodyLimit，handler 和提取器都不运行length limit exceeded
没有 Content-Length（或它没超），读的过程中超额axum 的提取器把错误变成 rejectionFailed to buffer the request body: length limit exceeded
handler 自己读 Request 的 body你的 handler，拿到的是一个 body 错误由你决定
  • 第二、三行和第五行状态码、正文完全一样，但触发的是不同的上限：前两个是 tower-http 的 4 KiB，第五行是 axum 默认的 2 MiB（tower-http 给了 4 MiB，Bytes 提取器又在外面套了一层 2 MiB 的 Limited）。两个 Limited 嵌套，额度小的那个先触发；axum-core 的 FailedToBufferBody::from_err 拆开错误往下找 http_body_util::LengthLimitError，哪一层产生的都能找到，所以都是 413，从响应上分不出来。
  • 第三行：Content-Length 说只有 100 字节，额度就被收紧到 100。真实连接上 hyper 按 Content-Length 截断 body，不会出现「说 100 发 8 KiB」，这里只是演示额度取的是较小值。
  • 想要超过 2 MiB 的上传，光加大 RequestBodyLimitLayer 不够，还要 DefaultBodyLimit::disable() 或 DefaultBodyLimit::max(..)，最后两行就是这样。axum-core 里 DefaultBodyLimit::disable 的文档示例就是 .layer(DefaultBodyLimit::disable()) 再 .layer(RequestBodyLimitLayer::new(10 * 1000 * 1000))。

CorsLayer：预检请求不进 handler

Cors::call：

    fn call(&mut self, req: Request<ReqBody>) -> Self::Future {
        let (parts, body) = req.into_parts();
        let origin = parts.headers.get(&header::ORIGIN);

        let mut headers = HeaderMap::new();

        // These headers are applied to both preflight and subsequent regular CORS requests:
        // https://fetch.spec.whatwg.org/#http-responses

        headers.extend(self.layer.allow_credentials.to_header(origin, &parts));
        headers.extend(self.layer.allow_private_network.to_header(origin, &parts));
        headers.extend(self.layer.vary.to_header());

        let allow_origin_future = self.layer.allow_origin.to_future(origin, &parts);

        // Return results immediately upon preflight request
        if parts.method == Method::OPTIONS {
            // These headers are applied only to preflight requests
            headers.extend(self.layer.allow_methods.to_header(&parts));
            headers.extend(self.layer.allow_headers.to_header(&parts));
            headers.extend(self.layer.max_age.to_header(origin, &parts));

            ResponseFuture {
                inner: Kind::PreflightCall {
                    allow_origin_future,
                    headers,
                },
            }
        } else {
            // This header is applied only to non-preflight requests
            headers.extend(self.layer.expose_headers.to_header(&parts));

            let req = Request::from_parts(parts, body);
            ResponseFuture {
                inner: Kind::CorsCall {
                    allow_origin_future,
                    allow_origin_complete: false,
                    future: self.inner.call(req),
                    headers,
                },
            }
        }
    }
  • 所有 OPTIONS 请求都被当成预检，不看有没有 Origin、Access-Control-Request-Method，内层永远不会被调用。响应是 Response::new(B::default())：200、空 body，加上算好的 CORS 头。你给 OPTIONS 注册的 handler 在这一层后面就成了摆设。
  • 允许与否不靠状态码。来源不在允许列表里时，只是不写 Access-Control-Allow-Origin，响应照样是 200，非预检请求的 handler 照样执行。拦截由浏览器做：它看不到匹配的头，就不把响应交给页面的脚本。所以 CORS 不是访问控制，curl 不受它影响。
  • 允许的方法、头不按请求过滤：allow_methods([GET, POST]) 对任何预检都回 GET,POST，由浏览器比对。
  • allow_origin 给一个 HeaderValue（常量）时，对谁都回这个值，不看请求的 Origin；给一个列表或谓词时，匹配上才把请求的 Origin 原样写回去，这时响应随 Origin 而变，Vary: origin 才有意义。0.7.0 起 Vary 由配置算出来（update_vary_header），见下面的冷知识。
  • 非法组合在装配时就 panic：ensure_usable_cors_rules 在 Layer::layer 里执行，allow_credentials(true) 不能和任何一个 *（Any）同时出现，这是 Fetch 规范的规定。
use axum::{
  Router,
  body::Body,
  http::{HeaderValue, Method, Request, header},
  routing::get,
};
use std::time::Duration;
use tower::ServiceExt;
use tower_http::cors::{Any, CorsLayer};

async fn send(app: &Router, name: &str, method: Method, path: &str, headers: &[(&str, &str)]) {
  let mut req = Request::builder().method(method).uri(path);
  for (k, v) in headers {
    req = req.header(*k, *v);
  }
  let res = app.clone().oneshot(req.body(Body::empty()).unwrap()).await.unwrap();
  println!("{name} -> {}", res.status());
  let mut hs: Vec<_> = res.headers().iter().map(|(k, v)| format!("{k}: {}", v.to_str().unwrap())).collect();
  hs.sort();
  for h in hs {
    println!("    {h}");
  }
}

#[tokio::main]
async fn main() {
  let cors = CorsLayer::new()
    .allow_origin([HeaderValue::from_static("https://app.example")])
    .allow_methods([Method::GET, Method::POST])
    .allow_headers([header::CONTENT_TYPE])
    .max_age(Duration::from_secs(600));
  let app = Router::new()
    .route("/api", get(|| async { "data" }).options(|| async { println!("    (options handler ran)"); "options" }))
    .layer(cors);

  let pre = [("origin", "https://app.example"), ("access-control-request-method", "POST"), ("access-control-request-headers", "content-type")];
  send(&app, "OPTIONS /api preflight", Method::OPTIONS, "/api", &pre).await;
  send(&app, "OPTIONS /api from evil.example", Method::OPTIONS, "/api", &[("origin", "https://evil.example"), ("access-control-request-method", "DELETE")]).await;
  send(&app, "OPTIONS /api no Origin", Method::OPTIONS, "/api", &[]).await;
  send(&app, "OPTIONS /nope", Method::OPTIONS, "/nope", &pre[..2]).await;
  send(&app, "GET /api from app.example", Method::GET, "/api", &[("origin", "https://app.example")]).await;
  send(&app, "GET /api from evil.example", Method::GET, "/api", &[("origin", "https://evil.example")]).await;

  println!("-- allow_origin(one HeaderValue)");
  let app1 = Router::new()
    .route("/api", get(|| async { "data" }))
    .layer(CorsLayer::new().allow_origin(HeaderValue::from_static("https://app.example")));
  send(&app1, "GET /api from evil.example", Method::GET, "/api", &[("origin", "https://evil.example")]).await;

  println!("-- permissive()");
  let app2 = Router::new().route("/api", get(|| async { "data" })).layer(CorsLayer::permissive());
  send(&app2, "OPTIONS /api preflight", Method::OPTIONS, "/api", &pre).await;

  println!("-- credentials + Any");
  std::panic::set_hook(Box::new(|_| {}));
  let r = std::panic::catch_unwind(|| {
    let _: Router = Router::new()
      .route("/api", get(|| async { "data" }))
      .layer(CorsLayer::new().allow_credentials(true).allow_origin(Any));
  });
  let msg = r.unwrap_err();
  println!("panicked at .layer(): {}", msg.downcast_ref::<String>().cloned().or_else(|| msg.downcast_ref::<&str>().map(|s| s.to_string())).unwrap());
}

OPTIONS /api preflight -> 200 OK
    access-control-allow-headers: content-type
    access-control-allow-methods: GET,POST
    access-control-allow-origin: https://app.example
    access-control-max-age: 600
    content-length: 0
    vary: origin
OPTIONS /api from evil.example -> 200 OK
    access-control-allow-headers: content-type
    access-control-allow-methods: GET,POST
    access-control-max-age: 600
    content-length: 0
    vary: origin
OPTIONS /api no Origin -> 200 OK
    access-control-allow-headers: content-type
    access-control-allow-methods: GET,POST
    access-control-max-age: 600
    content-length: 0
    vary: origin
OPTIONS /nope -> 200 OK
    access-control-allow-headers: content-type
    access-control-allow-methods: GET,POST
    access-control-allow-origin: https://app.example
    access-control-max-age: 600
    vary: origin
GET /api from app.example -> 200 OK
    access-control-allow-origin: https://app.example
    content-length: 4
    content-type: text/plain; charset=utf-8
    vary: origin
GET /api from evil.example -> 200 OK
    content-length: 4
    content-type: text/plain; charset=utf-8
    vary: origin
-- allow_origin(one HeaderValue)
GET /api from evil.example -> 200 OK
    access-control-allow-origin: https://app.example
    content-length: 4
    content-type: text/plain; charset=utf-8
-- permissive()
OPTIONS /api preflight -> 200 OK
    access-control-allow-headers: *
    access-control-allow-methods: *
    access-control-allow-origin: *
    allow: GET,HEAD
    content-length: 0
-- credentials + Any
panicked at .layer(): Invalid CORS configuration: Cannot combine `Access-Control-Allow-Credentials: true` with `Access-Control-Allow-Origin: *`
  • (options handler ran) 一次也没打印。
  • OPTIONS /nope：一个不存在的路径，预检也是 200。Router::layer 连 fallback 一起套（layer 与 route_layer），CORS 在 fallback 的 404 之前就回了。
  • permissive() 那一段多了 allow: GET,HEAD：/api 没有注册 options，OPTIONS 落到 MethodRouter 的 405 fallback，CORS 套在这个 fallback 外面，抢先回了 200；但 axum 的 RouteFuture 只要走了 fallback 就补 Allow 头，不看状态码（MethodRouter）。
  • content-length: 0 同样是 RouteFuture 补的；/nope 走的是 Router 级别的 fallback，没有这一步。
📝NOTE
冷知识：Vary 以前总是三个头

0.2.1 的 #199 让 CorsLayer 加上 Vary，此后不管怎么配置，它都写 origin, access-control-request-method, access-control-request-headers。issue #539「CORS module excessively sets the Vary response header」指出：「cors::CorsLayer should only add a request header to the Vary response header if a dynamic response is calculated based on that request header. .permissive(), for example, does not dynamically adjust the response at all.」多余的 Vary 会让缓存按根本不影响响应的头拆分条目。

0.7.0 的 #674「Relaxed Vary header defaults」改成在 Layer::layer 里调 update_vary_header：问 allow_origin、allow_methods、allow_headers 各自是否随请求变化，只把会变的那几个写进 Vary。方法和头只有两种内部表示：常量（Any、列表都算）和 mirror_request，只有后者会变；来源多两种：列表和谓词，它们会把请求的 Origin 写回去，也算会变。所以上面的列表配置只有 vary: origin，常量和 permissive() 没有 Vary。调用过 .vary(...) 的不受影响（is_vary_custom）。CHANGELOG 记为「cors: relax the Vary header defaults」。

TraceLayer：五个回调

TraceLayer::new_for_http 用 ServerErrorsAsFailures 做分类器，五个回调都用默认实现：on_request、on_response、on_eos 记 DEBUG 级别的事件，on_failure 记 ERROR，on_body_chunk 什么都不做（mod.rs 里的 DEFAULT_MESSAGE_LEVEL 和 DEFAULT_ERROR_LEVEL）。每个回调都可以换成闭包。它们在三个地方被调用：

  1. Trace::call：记下开始时间，make_span 造一个 span，进入 span 调 on_request，再调内层。
  2. ResponseFuture::poll：内层返回响应时，先分类，再调 on_response，分类结果是失败就接着调 on_failure；内层返回 Err 时只调 on_failure。然后把 body 包成 trace::ResponseBody。
  3. ResponseBody::poll_frame：每个数据帧调 on_body_chunk；body 结束（None、trailers，或者 is_end_stream() 为真）时调 on_eos。

分类器 ServerErrorsAsFailures 只看状态码：

    fn classify_response<B>(
        self,
        res: &Response<B>,
    ) -> ClassifiedResponse<Self::FailureClass, Self::ClassifyEos> {
        if res.status().is_server_error() {
            ClassifiedResponse::Ready(Err(ServerErrorsFailureClass::StatusCode(res.status())))
        } else {
            ClassifiedResponse::Ready(Ok(()))
        }
    }

5xx 是失败，4xx 不是（客户端的错不算服务的故障）。它总是返回 Ready，ClassifyEos 是 NeverClassifyEos，意思是看完响应头就下结论，不等 body。gRPC 用的 GrpcErrorsAsFailures 才会返回 RequiresEos，等 trailers 里的 grpc-status。

在 axum 里，Router 的错误类型是 Infallible，ResponseFuture 的 Err 分支永远走不到，on_failure 只会因为 5xx 被调用。下面把 CatchPanicLayer 放在 TraceLayer 里面（后 .layer 的在外层），看几种响应各触发了哪些回调：

use axum::{
  Router,
  body::{Body, Bytes},
  http::{Request, StatusCode},
  routing::get,
};
use http_body_util::BodyExt;
use std::time::Duration;
use tower::ServiceExt;
use tower_http::{
  catch_panic::CatchPanicLayer,
  classify::ServerErrorsFailureClass,
  trace::TraceLayer,
};
use tracing::Span;

#[tokio::main]
async fn main() {
  let trace = TraceLayer::new_for_http()
    .on_request(|req: &Request<Body>, _: &Span| println!("  on_request {} {}", req.method(), req.uri()))
    .on_response(|res: &axum::response::Response, _: Duration, _: &Span| println!("  on_response {}", res.status()))
    .on_body_chunk(|chunk: &Bytes, _: Duration, _: &Span| println!("  on_body_chunk {} bytes", chunk.len()))
    .on_eos(|_: Option<&axum::http::HeaderMap>, _: Duration, _: &Span| println!("  on_eos"))
    .on_failure(|class: ServerErrorsFailureClass, _: Duration, _: &Span| println!("  on_failure {class}"));

  let app = Router::new()
    .route("/ok", get(|| async { "hello" }))
    .route("/teapot", get(|| async { StatusCode::IM_A_TEAPOT }))
    .route("/boom", get(|| async { (StatusCode::INTERNAL_SERVER_ERROR, "boom") }))
    .route("/panic", get(|| async {
      if true {
        panic!("handler blew up");
      }
      "unreachable"
    }))
    .route(
      "/body-error",
      get(|| async {
        let chunks = futures::stream::iter([
          Ok(Bytes::from_static(b"part")),
          Err(std::io::Error::other("disk gone")),
        ]);
        Body::from_stream(chunks)
      }),
    )
    .layer(CatchPanicLayer::new())
    .layer(trace);

  std::panic::set_hook(Box::new(|info| println!("  (panic hook: {})", info.payload().downcast_ref::<&str>().unwrap())));
  for path in ["/ok", "/teapot", "/boom", "/panic", "/body-error"] {
    println!("GET {path}");
    let res = app.clone().oneshot(Request::get(path).body(Body::empty()).unwrap()).await.unwrap();
    match res.into_body().collect().await {
      Ok(c) => println!("  client got {:?}", String::from_utf8_lossy(&c.to_bytes())),
      Err(e) => println!("  client got error: {e}"),
    }
  }
}

GET /ok
  on_request GET /ok
  on_response 200 OK
  on_body_chunk 5 bytes
  on_eos
  client got "hello"
GET /teapot
  on_request GET /teapot
  on_response 418 I'm a teapot
  on_eos
  client got ""
GET /boom
  on_request GET /boom
  on_response 500 Internal Server Error
  on_failure Status code: 500 Internal Server Error
  on_body_chunk 4 bytes
  on_eos
  client got "boom"
GET /panic
  on_request GET /panic
  (panic hook: handler blew up)
  on_response 500 Internal Server Error
  on_failure Status code: 500 Internal Server Error
  on_body_chunk 16 bytes
  on_eos
  client got "Service panicked"
GET /body-error
  on_request GET /body-error
  on_response 200 OK
  on_body_chunk 4 bytes
  client got error: disk gone
  • 5xx 时 on_response 在 on_failure 前面，两个都调。
  • /panic：CatchPanicLayer 在里面把 panic 变成了 500，Trace 看到的就是一个普通的 500。反过来摆（CatchPanicLayer 在外），panic 会穿过 Trace 的 future，Trace 什么都记不下。
  • /body-error：body 读到一半出错，on_failure 没有被调用，on_eos 也没有。原因在 trace/body.rs 的错误分支：它要同时拿到 classify_eos 和 on_failure 才调用，而 ServerErrorsAsFailures 在响应头阶段就给出了 Ready，classify_eos 是 None。用 new_for_http 时，响应头已经发出之后的失败，Trace 的默认配置看不见。需要的话，给 body 错误自己记日志，或者在 on_body_chunk / on_eos 里判断「有没有正常结束」。

0.7.0 还有一处和日志有关的改动：默认回调的事件现在显式挂在请求的 span 下，而不是依赖调用时的当前 span（#690，「This fixes intermittent cases where events could appear without their request span attached」）。

CatchPanicLayer：只管到响应头为止

CatchPanic::call 用了两次 catch_unwind：

    fn call(&mut self, req: Request<ReqBody>) -> Self::Future {
        match std::panic::catch_unwind(AssertUnwindSafe(|| self.inner.call(req))) {
            Ok(future) => ResponseFuture {
                kind: Kind::Future {
                    future: AssertUnwindSafe(future).catch_unwind(),
                    panic_handler: Some(self.panic_handler.clone()),
                },
            },
            Err(panic_err) => ResponseFuture {
                kind: Kind::Panicked {
                    panic_err: Some(panic_err),
                    panic_handler: Some(self.panic_handler.clone()),
                },
            },
        }
    }

一次包住同步的 inner.call，一次用 futures 的 FutureExt::catch_unwind 包住返回的 future。axum 的 handler 是 async 函数，panic 发生在 future 被 poll 的时候，由第二次捕获。捕获到之后交给 ResponseForPanic，默认的 DefaultResponseForPanic 用 tracing::error! 记下 panic 信息（能 downcast 成 String 或 &str 时），返回 500、Content-Type: text/plain; charset=utf-8、正文 Service panicked。可以用 CatchPanicLayer::custom(|err| ...) 换成自己的响应。

响应 body 只做了类型擦除（UnsyncBoxBody），没有包 catch_unwind，所以 body 被 poll 时的 panic 不归它管：

use axum::{
  Router,
  body::{Body, Bytes},
  http::Request,
  routing::get,
};
use futures::StreamExt;
use http_body_util::BodyExt;
use tower::ServiceExt;
use tower_http::catch_panic::CatchPanicLayer;

fn app() -> Router {
  Router::new()
    .route("/handler", get(|| async {
      if true {
        panic!("in handler");
      }
      "unreachable"
    }))
    .route(
      "/body",
      get(|| async {
        let s = futures::stream::iter([1, 2]).map(|i| {
          if i == 2 {
            panic!("in body stream");
          }
          Ok::<_, std::io::Error>(Bytes::from_static(b"first"))
        });
        Body::from_stream(s)
      }),
    )
}

async fn hit(app: Router, path: &'static str) -> String {
  let res = app.oneshot(Request::get(path).body(Body::empty()).unwrap()).await.unwrap();
  let status = res.status();
  let ct = res.headers().get("content-type").cloned();
  let body = res.into_body().collect().await.unwrap().to_bytes();
  format!("{status} {ct:?} {:?}", String::from_utf8_lossy(&body))
}

#[tokio::main]
async fn main() {
  std::panic::set_hook(Box::new(|_| {}));
  for (name, app) in [("without CatchPanicLayer", app()), ("with CatchPanicLayer", app().layer(CatchPanicLayer::new()))] {
    println!("-- {name}");
    for path in ["/handler", "/body"] {
      match tokio::spawn(hit(app.clone(), path)).await {
        Ok(line) => println!("{path:<9} {line}"),
        Err(e) => {
          let p = e.into_panic();
          println!("{path:<9} task panicked: {}", p.downcast_ref::<&str>().unwrap());
        }
      }
    }
  }
}

-- without CatchPanicLayer
/handler  task panicked: in handler
/body     task panicked: in body stream
-- with CatchPanicLayer
/handler  500 Internal Server Error Some("text/plain; charset=utf-8") "Service panicked"
/body     task panicked: in body stream

没有这一层时，panic 沿着 future 一直传到驱动它的任务；在真实服务器上，那是 hyper 驱动这条连接的任务，tokio 捕获任务的 panic，连接随任务一起结束，客户端读到 EOF、拿不到任何响应字节；服务器本身照常接受新连接。有了这一层，handler 里的 panic 变成 500；但 /body 的 panic 发生在响应头已经交出去、body 被读取的时候，CatchPanicLayer 早就完成了它的 future，照样传到任务里。

CsrfLayer

0.7.0 新增的 CsrfLayer（#699）移植了 Go 1.25 的跨站请求伪造防护：不用 token，凭 Sec-Fetch-Site 和 Origin 判断，GET / HEAD / OPTIONS 一律放行，其余方法来自不受信任的跨站来源时回 403。

❗IMPORTANT
动脑筋：同时到期
use axum::{Router, body::Body, http::{Request, StatusCode}, routing::get};
use std::time::Duration;
use tokio::time::{Instant, sleep};
use tower::ServiceExt;
use tower_http::timeout::TimeoutLayer;

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
  let app = Router::new()
    .route("/", get(|| async {
      sleep(Duration::from_millis(100)).await;
      "done"
    }))
    .layer(TimeoutLayer::with_status_code(StatusCode::SERVICE_UNAVAILABLE, Duration::from_millis(100)));
  let start = Instant::now();
  let res = app.oneshot(Request::get("/").body(Body::empty()).unwrap()).await.unwrap();
  println!("{} at {} ms", res.status(), start.elapsed().as_millis());
}
暂停时钟下，handler 睡 100 毫秒，超时也是 100 毫秒，两个定时器在同一时刻到期。打印 200 还是 503？

动脑筋答案

503 Service Unavailable at 100 ms

跑 3 次相同。时钟暂停时，call 创建超时的 Sleep 和 handler 第一次被 poll 时创建自己的 Sleep，中间虚拟时间没有前进，两个截止时间都是 100 毫秒。时钟推进到 100 毫秒，两个定时器同时触发、同一个任务被唤醒。ResponseFuture::poll 先 poll sleep，它已就绪，直接返回 503，内层 future 连 poll 的机会都没有。换成 tower 的 Timeout（先 poll 内层），同样的情形会拿到 200。真实时钟下两个截止时间几乎不会完全相等，这里只是把「谁先 poll」这件事单独拿出来看。

小结

  • TimeoutLayer：在 call 时开始计时，先查超时再 poll 内层；超时回一个空 body 的响应，状态码自己选（new 的 408 已 deprecated）；读请求 body 算在内，响应 body 不算，body 的超时另有四个 layer。
  • RequestBodyLimitLayer：Content-Length 超限当场 413（length limit exceeded），否则只给 body 套 Limited；和 axum 的 DefaultBodyLimit 嵌套时额度小的先触发，axum 提取器把两者都报成同一个 413；要放大上传上限，两个都要调。
  • CorsLayer：所有 OPTIONS 都在这一层直接回 200，handler 不运行；不允许的来源只是少一个头，状态码不变；0.7.0 起 Vary 按配置计算；非法组合在 .layer 时 panic。
  • TraceLayer：on_request 在 call，on_response / on_failure 在响应头出来时，on_body_chunk / on_eos 在 body 流动时；new_for_http 只把 5xx 算失败，body 中途出错不会触发 on_failure。
  • CatchPanicLayer：捕获 call 和响应 future 里的 panic，换成 500 Service panicked；body 里的 panic 不管。要让 Trace 看到它产生的 500，把它放在 Trace 里面。
  • CsrfLayer（0.7.0）：凭 Sec-Fetch-Site / Origin 拒绝跨站的写请求，回 403。

TraceLayer 的类型参数、分类器怎么决定 body 出错算不算失败，见 tower-http 卷 Trace 的内部 和 classify；CatchPanicLayer 的两次捕获与 panic 载荷见 CatchPanic；CorsLayer、TimeoutLayer 为什么要求内层 body 实现 Default，见 四件套。

💬 关联讨论

发起讨论
加载中...