优雅停机一直不结束

主线位置：服务端主线的终点。停机时每条连接的 Connection future 被告知「做完手上的请求就结束」，停机的一方等这些 future 全部结束。一条连接「手上的请求」如果永远做不完，等待就永远不结束；反过来，已经升级、离开了 Connection future 的连接，根本不在等待的名单上。

flowchart TB
  sig["停机信号"] --> idle["空闲连接：立刻关"]
  sig --> busy["普通请求：做完再关"]
  sig --> sse["SSE：body 永不结束"]
  sig --> upg["升级后的连接"]
  sse --> hang["shutdown 一直等"]
  upg --> gone["早已不在名单上"]
  gone --> cut["进程退出时直接断开"]
  hang --> timeout["只能靠超时"]

单条连接的 graceful_shutdown 做什么、hyper-util 的 GracefulShutdown 怎么用一个 watch 通道通知和等待、axum 怎么用两个 watch 通道做同一件事，都在优雅停机一章。这一页只看两种长连接：一个永不结束的 SSE 响应，一个升级成裸字节流的连接（WebSocket 也是这样升级的）。

复现一：GracefulShutdown

auto::Builder 加 serve_connection_with_upgrades 接两条连接，都交给同一个 GracefulShutdown：

sse：响应 body 每 100 ms 发一块，永不结束；
upgrade：服务端回 101，在新任务里 hyper::upgrade::on 拿到 IO，每 100 ms 写一行 tick。

300 ms 后调 shutdown()，套 1 秒超时。程序带参数 fix 时启用后面「修法」一节的改动，这里先看不带参数的运行。

hyper = { version = "=1.12.0", features = ["client", "server", "http1", "http2"] }
hyper-util = { version = "=0.1.21", features = ["server-auto", "server-graceful", "tokio"] }
http-body-util = "=0.1.5"
bytes = "1"
futures = "0.3"
tokio = { version = "=1.54.1", features = ["full"] }

use std::convert::Infallible;
use std::sync::atomic::{AtomicUsize, Ordering::SeqCst};
use std::sync::OnceLock;
use std::time::{Duration, Instant};

use bytes::Bytes;
use futures::StreamExt as _;
use http_body_util::{combinators::BoxBody, BodyExt, Empty, StreamBody};
use hyper::body::{Frame, Incoming};
use hyper::client::conn::http1 as client_h1;
use hyper::service::service_fn;
use hyper::{Request, Response, StatusCode};
use hyper_util::rt::{TokioExecutor, TokioIo};
use hyper_util::server::conn::auto;
use hyper_util::server::graceful::GracefulShutdown;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::watch;

type B = BoxBody<Bytes, Infallible>;

static START: OnceLock<Instant> = OnceLock::new();
static SSE_CHUNKS: AtomicUsize = AtomicUsize::new(0); // seen by the SSE client
static TICKS: AtomicUsize = AtomicUsize::new(0); // seen by the upgrade client

// milliseconds since shutdown started, or "before"
fn ms() -> String {
  START.get().map(|t| format!("{} ms", t.elapsed().as_millis())).unwrap_or("before".into())
}
fn counts() -> String {
  format!("sse chunks {}, upgrade ticks {}", SSE_CHUNKS.load(SeqCst), TICKS.load(SeqCst))
}

async fn handler(
  mut req: Request<Incoming>,
  stop: watch::Receiver<bool>,
  upgrades: watch::Receiver<()>,
  fix: bool,
) -> Result<Response<B>, Infallible> {
  if req.uri().path() == "/upgrade" {
    // answer 101, then talk raw bytes on the upgraded IO in a new task
    tokio::spawn(async move {
      let _guard = upgrades; // dropped when this task ends
      let mut io = TokioIo::new(hyper::upgrade::on(&mut req).await.unwrap());
      let mut stop = stop;
      loop {
        tokio::select! {
          _ = tokio::time::sleep(Duration::from_millis(100)) => {}
          _ = stop.wait_for(|v| *v), if fix => break,
        }
        if io.write_all(b"tick\n").await.is_err() {
          break;
        }
      }
      let _ = io.shutdown().await;
      println!("{:>7} server: upgraded task ended", ms());
    });
    let res = Response::builder()
      .status(StatusCode::SWITCHING_PROTOCOLS)
      .header("connection", "upgrade")
      .header("upgrade", "demo")
      .body(Empty::new().boxed())
      .unwrap();
    return Ok(res);
  }
  // /sse: one chunk every 100 ms, forever
  let ticks = futures::stream::unfold((), |()| async {
    tokio::time::sleep(Duration::from_millis(100)).await;
    Some((Ok::<_, Infallible>(Frame::data(Bytes::from("data: x\n\n"))), ()))
  });
  if fix {
    let mut stop = stop;
    let signal = async move {
      let _ = stop.wait_for(|v| *v).await;
    };
    return Ok(Response::new(BodyExt::boxed(StreamBody::new(ticks.take_until(signal)))));
  }
  Ok(Response::new(BodyExt::boxed(StreamBody::new(ticks))))
}

async fn sse_client(addr: std::net::SocketAddr) {
  let stream = TcpStream::connect(addr).await.unwrap();
  let (mut tx, conn) = client_h1::handshake::<_, Empty<Bytes>>(TokioIo::new(stream)).await.unwrap();
  tokio::spawn(conn);
  let res = tx.send_request(Request::get("/sse").body(Empty::new()).unwrap()).await.unwrap();
  let mut body = res.into_body();
  let end = loop {
    match body.frame().await {
      Some(Ok(_)) => SSE_CHUNKS.fetch_add(1, SeqCst),
      Some(Err(e)) => break format!("error: {e}"),
      None => break "end of body".to_string(),
    };
  };
  println!("{:>7} client: sse body {end}", ms());
}

async fn upgrade_client(addr: std::net::SocketAddr) {
  let mut s = TcpStream::connect(addr).await.unwrap();
  s.write_all(b"GET /upgrade HTTP/1.1\r\nhost: x\r\nconnection: upgrade\r\nupgrade: demo\r\n\r\n").await.unwrap();
  let mut buf = [0u8; 1024];
  loop {
    match s.read(&mut buf).await {
      Ok(0) | Err(_) => break,
      Ok(n) => TICKS.fetch_add(String::from_utf8_lossy(&buf[..n]).matches("tick").count(), SeqCst),
    };
  }
  println!("{:>7} client: upgrade stream closed", ms());
}

#[tokio::main]
async fn main() {
  let fix = std::env::args().nth(1).as_deref() == Some("fix");
  let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
  let addr = listener.local_addr().unwrap();
  let graceful = GracefulShutdown::new();
  let (stop_tx, stop_rx) = watch::channel(false);
  let (upgrades_tx, upgrades_rx) = watch::channel(()); // counts upgraded tasks
  let builder = auto::Builder::new(TokioExecutor::new());

  // accept exactly two connections: first the SSE client, then the upgrade client
  tokio::spawn(sse_client(addr));
  for label in ["sse", "upgrade"] {
    if label == "upgrade" {
      tokio::spawn(upgrade_client(addr));
    }
    let (stream, _) = listener.accept().await.unwrap();
    let (stop, upgrades) = (stop_rx.clone(), upgrades_rx.clone());
    let svc = service_fn(move |req| handler(req, stop.clone(), upgrades.clone(), fix));
    let conn = builder.serve_connection_with_upgrades(TokioIo::new(stream), svc);
    let conn = graceful.watch(conn.into_owned());
    tokio::spawn(async move {
      let r = conn.await.map_err(|e| e.to_string());
      println!("{:>7} server: {label} connection future ended {r:?}", ms());
    });
  }

  tokio::time::sleep(Duration::from_millis(300)).await;
  println!("fix = {fix}; count() = {}; {}", graceful.count(), counts());
  START.set(Instant::now()).unwrap();
  let _ = stop_tx.send(true); // nobody listens unless fix
  match tokio::time::timeout(Duration::from_secs(1), graceful.shutdown()).await {
    Ok(()) => println!("{:>7} shutdown() returned; {}", ms(), counts()),
    Err(_) => println!("{:>7} shutdown() timed out; {}", ms(), counts()),
  }
  drop(upgrades_rx);
  // upgraded connections are not watched by GracefulShutdown; wait for them here
  match tokio::time::timeout(Duration::from_secs(1), upgrades_tx.closed()).await {
    Ok(()) => println!("{:>7} upgraded tasks: all ended", ms()),
    Err(_) => println!("{:>7} upgraded tasks: still waiting, gave up", ms()),
  }
  tokio::time::sleep(Duration::from_millis(50)).await; // let the clients report
  println!("{:>7} main returns; {}", ms(), counts());
}

M2 Max、rustc 1.99.0、--release，不带参数，其中一次的输出：

 before server: upgrade connection future ended Ok(())
fix = false; count() = 1; sse chunks 2, upgrade ticks 2
1002 ms shutdown() timed out; sse chunks 12, upgrade ticks 12
2003 ms upgraded tasks: still waiting, gave up
2054 ms main returns; sse chunks 23, upgrade ticks 23

连跑 3 次，行和顺序相同，毫秒数差 2 ms 以内，最后一行的计数是 22 或 23。

count() = 1：watch 了两条连接，停机前只剩一个接收端。第一行说明，升级那条连接的 Connection future 早在停机前就以 Ok(()) 结束了。
shutdown() 超时：SSE 连接已经收到 graceful_shutdown，但它的 body 永远写不完，接收端一直活着。
两个计数一直在涨：停机期间（0 到 1002 ms）SSE 又发了 10 块，升级后的连接也写了 10 行；超时之后照样在涨，直到 main 返回、运行时把所有任务 drop 掉。升级后的那条连接从头到尾没有收到任何停机通知。

为什么：一个做不完，一个早就走了

SSE。 HTTP/1 的 graceful_shutdown 只是关掉 keep-alive：手上的响应照常写完，带上 connection: close，然后结束。hyper 不知道这个 body 是 SSE，它只看到一个还没结束的 body。HTTP/2 也一样，GOAWAY 只拒绝新的流，已经开始的流要等它自己结束（HTTP/2 服务端的停机实验）。所以这类连接在 hyper 的语义里就是「当前请求还没做完」，等多久都是对的。

升级。 UpgradeableConnection 的 poll 在 Dispatcher 报告 Dispatched::Upgrade 时，把 IO 和读缓冲里剩下的字节交给 upgrade::on 那一头，然后自己结束（auto 的升级入口在 HTTP/1 下用的就是它）：

                Ok(proto::Dispatched::Upgrade(pending)) => {
                    let (io, buf, _) = self
                        .inner
                        .take()
                        .expect("upgradeable server connection missing after upgrade")
                        .conn
                        .into_inner();
                    pending.fulfill(Upgraded::new(io, buf));
                    Poll::Ready(Ok(()))
                }

future 一结束，GracefulShutdown 包在外面的接收端就被 drop，计数减一。之后这条 TCP 连接活在你自己 spawn 的任务里，hyper、hyper-util 都不再持有它，没有办法通知它，也不会等它。UpgradeableConnection::graceful_shutdown 的注释写得很直白：「Connection (inner) is None if it was upgraded (and poll is Ready). In that case, we don't need to call graceful_shutdown.」

所以两种长连接的后果正好相反：SSE 让停机等不完；升级后的连接让停机不等，停机「成功」返回，进程退出时它被直接切断，对端收不到任何关闭消息（WebSocket 的 Close 帧要你自己的代码去发）。

📝NOTE

冷知识：这个 issue 从 hyper 0.14 开到现在

hyper 的 issue #2787「Graceful shutdown never stops the server when there is an open sse connection」是 2022 年用 hyper 0.14.17 加 axum 0.4.8 报的，到 v1.12.0 仍然开着。报告里顺带提了另一半现象：「Note that with websocket that issue doesn't exist, the server shuts down normally closing the ws connections.」回复解释了原因：「when a regular HTTP connection is upgraded to a WebSocket connection it leaves hyper's connection pool and is handed over to whatever WebSocket library your using ... So that means from hyper's perspective the connection is already gone with graceful shutdown is triggered.」也就是上一节 Dispatched::Upgrade 那段代码。

关于 SSE，回复给的办法是把流包一层，「Run a stream until it completes or we receive the shutdown signal」，并且认为 hyper 自己做不了：「Hyper doesn't know anything about SSE as it's just a long running response body.」从 0.14 到 1.12.0，Server 搬进了 hyper-util 和 axum，停机的实现换了两次，这一点没变。

复现二：axum 的 with_graceful_shutdown

axum 0.8.9 的连接任务（axum/src/serve/mod.rs 的 handle_connection）和上面是同一个结构：每条连接 tokio::spawn 一个任务，任务里用 select! 同时等连接和停机信号：

        let mut conn = pin!(builder.serve_connection_with_upgrades(io, hyper_service));
        let mut signal_closed = pin!(signal_tx.closed().fuse());

        loop {
            tokio::select! {
                result = conn.as_mut() => {
                    if let Err(_err) = result {
                        trace!("failed to serve connection: {_err:#}");
                    }
                    break;
                }
                _ = &mut signal_closed => {
                    trace!("signal received in task, starting graceful shutdown");
                    conn.as_mut().graceful_shutdown();
                }
            }
        }

        drop(close_rx);

close_rx 在连接 future 结束后才 drop，with_graceful_shutdown 返回的 future 最后等的是所有 close_rx 都没了。SSE 的连接 future 不结束，它就不返回；升级的连接 future 早就结束了，不算在里面。下面用 axum 重做一遍：同样的 /sse 和 /upgrade（axum 把 hyper 的 OnUpgrade 留在请求的 extensions 里，处理函数自己取出来），停机信号和超时都从同一个 watch 通道出发：

axum = "=0.8.9"
hyper = { version = "=1.12.0", features = ["client", "http1"] }
hyper-util = { version = "=0.1.21", features = ["tokio"] }
http-body-util = "=0.1.5"
bytes = "1"
futures = "0.3"
tokio = { version = "=1.54.1", features = ["full"] }

use std::convert::Infallible;
use std::sync::atomic::{AtomicUsize, Ordering::SeqCst};
use std::sync::OnceLock;
use std::time::{Duration, Instant};

use axum::body::Body;
use axum::extract::{Request, State};
use axum::http::StatusCode;
use axum::response::Response;
use axum::routing::get;
use axum::Router;
use bytes::Bytes;
use futures::StreamExt;
use http_body_util::{BodyExt, Empty};
use hyper_util::rt::TokioIo;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::watch;

static START: OnceLock<Instant> = OnceLock::new();
static SSE_CHUNKS: AtomicUsize = AtomicUsize::new(0);
static TICKS: AtomicUsize = AtomicUsize::new(0);

fn ms() -> String {
  START.get().map(|t| format!("{} ms", t.elapsed().as_millis())).unwrap_or("before".into())
}
fn counts() -> String {
  format!("sse chunks {}, upgrade ticks {}", SSE_CHUNKS.load(SeqCst), TICKS.load(SeqCst))
}

#[derive(Clone)]
struct AppState {
  stop: watch::Receiver<bool>,
  fix: bool,
}

async fn sse(State(st): State<AppState>) -> Body {
  let ticks = futures::stream::unfold((), |()| async {
    tokio::time::sleep(Duration::from_millis(100)).await;
    Some((Ok::<_, Infallible>(Bytes::from("data: x\n\n")), ()))
  });
  if st.fix {
    // end the body when the shutdown signal arrives
    let mut stop = st.stop;
    let signal = async move {
      let _ = stop.wait_for(|v| *v).await;
    };
    return Body::from_stream(ticks.take_until(signal));
  }
  Body::from_stream(ticks)
}

async fn upgrade(State(st): State<AppState>, mut req: Request) -> Response {
  // axum leaves hyper's OnUpgrade in the request extensions
  let on_upgrade = req.extensions_mut().remove::<hyper::upgrade::OnUpgrade>().unwrap();
  tokio::spawn(async move {
    let mut io = TokioIo::new(on_upgrade.await.unwrap());
    let mut stop = st.stop;
    loop {
      tokio::select! {
        _ = tokio::time::sleep(Duration::from_millis(100)) => {}
        _ = stop.wait_for(|v| *v), if st.fix => break,
      }
      if io.write_all(b"tick\n").await.is_err() {
        break;
      }
    }
    let _ = io.shutdown().await;
    println!("{:>7} server: upgraded task ended", ms());
  });
  Response::builder()
    .status(StatusCode::SWITCHING_PROTOCOLS)
    .header("connection", "upgrade")
    .header("upgrade", "demo")
    .body(Body::empty())
    .unwrap()
}

async fn sse_client(addr: std::net::SocketAddr) {
  let stream = TcpStream::connect(addr).await.unwrap();
  let (mut tx, conn) = hyper::client::conn::http1::handshake::<_, Empty<Bytes>>(TokioIo::new(stream)).await.unwrap();
  tokio::spawn(conn);
  let res = tx.send_request(hyper::Request::get("/sse").body(Empty::new()).unwrap()).await.unwrap();
  let mut body = res.into_body();
  let end = loop {
    match body.frame().await {
      Some(Ok(_)) => SSE_CHUNKS.fetch_add(1, SeqCst),
      Some(Err(e)) => break format!("error: {e}"),
      None => break "end of body".to_string(),
    };
  };
  println!("{:>7} client: sse body {end}", ms());
}

async fn upgrade_client(addr: std::net::SocketAddr) {
  let mut s = TcpStream::connect(addr).await.unwrap();
  s.write_all(b"GET /upgrade HTTP/1.1\r\nhost: x\r\nconnection: upgrade\r\nupgrade: demo\r\n\r\n").await.unwrap();
  let mut buf = [0u8; 1024];
  loop {
    match s.read(&mut buf).await {
      Ok(0) | Err(_) => break,
      Ok(n) => TICKS.fetch_add(String::from_utf8_lossy(&buf[..n]).matches("tick").count(), SeqCst),
    };
  }
  println!("{:>7} client: upgrade stream closed", ms());
}

#[tokio::main]
async fn main() {
  let fix = std::env::args().nth(1).as_deref() == Some("fix");
  let upgrade_only = std::env::args().nth(1).as_deref() == Some("upgrade-only");
  let (stop_tx, stop_rx) = watch::channel(false);
  let app = Router::new()
    .route("/sse", get(sse))
    .route("/upgrade", get(upgrade))
    .with_state(AppState { stop: stop_rx.clone(), fix });

  let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
  let addr = listener.local_addr().unwrap();
  let mut signal = stop_rx.clone();
  let serve = axum::serve(listener, app).with_graceful_shutdown(async move {
    let _ = signal.wait_for(|v| *v).await;
  });
  let mut server = tokio::spawn(async move {
    serve.await.unwrap();
    println!("{:>7} server: serve future returned", ms());
  });

  if !upgrade_only {
    tokio::spawn(sse_client(addr));
  }
  tokio::spawn(upgrade_client(addr));
  tokio::time::sleep(Duration::from_millis(300)).await;

  println!("fix = {fix}; {}", counts());
  START.set(Instant::now()).unwrap();
  stop_tx.send(true).unwrap();
  // the timeout starts when the signal is sent, not when the server starts
  if tokio::time::timeout(Duration::from_secs(1), &mut server).await.is_err() {
    println!("{:>7} still not finished, abort the serve task; {}", ms(), counts());
    server.abort();
  }
  tokio::time::sleep(Duration::from_millis(500)).await;
  println!("{:>7} main returns; {}", ms(), counts());
}

不带参数，连跑 3 次，行相同，毫秒数差 2 ms 以内：

fix = false; sse chunks 2, upgrade ticks 2
1003 ms still not finished, abort the serve task; sse chunks 12, upgrade ticks 12
1505 ms main returns; sse chunks 17, upgrade ticks 17

1 秒后 with_graceful_shutdown 还没返回，原因同上。值得注意的是第三行：abort 掉 serve 任务之后的 500 ms 里，SSE 和升级连接又各发了 5 次。axum 的每条连接是 handle_connection 里单独 tokio::spawn 的任务，serve future 只是在等它们；abort 掉它只是不再等，连接任务照样在运行时里跑。真正切断它们的是运行时关闭，也就是进程退出。

❗IMPORTANT

动脑筋：只有一条升级后的连接时，with_graceful_shutdown 多久返回？

用同一个程序，参数换成 upgrade-only：不连 SSE，只连 /upgrade，其余不变（不带 fix，升级任务不听停机信号）。server: serve future returned 会打印吗，在第几毫秒？main returns 那一行的 upgrade ticks 比停机前的 2 多还是一样？答案在本页末尾。

修法

三件事，缺一不可。

1. 超时兜底，然后真的退出。 超时要从发出停机信号开始算，不是从服务启动开始算。上面的 axum 程序就是这么做的：把 serve future spawn 出去拿到 JoinHandle，发信号，再对 JoinHandle 套 timeout。超时以后要让进程退出（从 main 返回，运行时关闭时 drop 所有任务）；只 abort serve 任务或者只放弃等待，连接任务还会继续跑。容器环境里，这个超时要比编排系统从 SIGTERM 到 SIGKILL 的宽限期短，否则进程会在超时之前被杀掉。

2. 让长响应自己听停机信号。 把 SSE 的流包一层 take_until(停机信号)：信号一来，流结束，body 正常收尾（chunked 的结束块写出去），HTTP/1 连接因为 keep-alive 已关而随之结束。客户端看到的是一个正常结束的响应，而不是被切断的连接。上面两个程序带 fix 参数时就是这样做的，SSE 的处理函数和停机共用同一个 watch 通道。

3. 自己记住升级出去的连接。 GracefulShutdown 和 axum 都管不到升级后的连接，要自己做两件事：通知它们（同一个停机信号，升级任务在 select! 里等它，收到后发 Close 帧、关闭写端），以及等它们结束。hyper-util 程序里用的是和 GracefulShutdown 一样的办法：每个升级任务持有一个 watch::Receiver<()> 的克隆（_guard），主函数 drop 掉自己那份后 upgrades_tx.closed().await，等所有克隆消失。tokio-util 的 TaskTracker 或者一个 JoinSet 也能做同一件事。

带 fix 参数，hyper-util 程序连跑 3 次，行相同，除最后一行（50 或 53 ms）外毫秒数都是 0：

 before server: upgrade connection future ended Ok(())
fix = true; count() = 1; sse chunks 2, upgrade ticks 2
   0 ms server: upgraded task ended
   0 ms server: sse connection future ended Ok(())
   0 ms shutdown() returned; sse chunks 2, upgrade ticks 2
   0 ms upgraded tasks: all ended
   0 ms client: upgrade stream closed
   0 ms client: sse body end of body
  53 ms main returns; sse chunks 2, upgrade ticks 2

axum 程序带 fix，连跑 4 次，0 ms 的几行都是 0 ms，两个 client 行的先后会变，最后一行 501 到 503 ms：

fix = true; sse chunks 2, upgrade ticks 2
   0 ms server: upgraded task ended
   0 ms server: serve future returned
   0 ms client: sse body end of body
   0 ms client: upgrade stream closed
 503 ms main returns; sse chunks 2, upgrade ticks 2

两边都在 0 ms 收尾：SSE 客户端看到的是 end of body，不是错误；升级连接被服务端主动关闭；计数停在停机前的 2。axum 程序里升级任务这次恰好先结束了，但 axum 并不等它（见下面的动脑筋答案）；要在 axum 里也等升级任务结束，得像 hyper-util 程序那样自己计数。

动脑筋答案

参数 upgrade-only，连跑 3 次，行相同，最后一行 502 或 503 ms：

fix = false; sse chunks 0, upgrade ticks 2
   0 ms server: serve future returned
 502 ms main returns; sse chunks 0, upgrade ticks 7

0 ms 就返回了。唯一的一条连接在升级时就离开了 axum 的连接任务，那个任务早已结束、close_rx 早已 drop，with_graceful_shutdown 没有任何东西要等。
upgrade ticks 从 2 涨到 7。停机「完成」之后，升级连接上的任务还在每 100 ms 写一行，没人通知它。进程这时退出，它会被直接切断；对一个真的 WebSocket 来说，客户端收不到 Close 帧，只看到连接断了。

小结

HTTP/1 的 graceful_shutdown 是「做完当前响应再关」，HTTP/2 是「拒绝新流、等在途的流」。SSE、长轮询这类永不结束的响应让 GracefulShutdown::shutdown() 和 axum 的 with_graceful_shutdown 永远等下去（hyper issue #2787，至今未关）。
升级后的连接不会拖住停机：UpgradeableConnection 交出 IO 后 future 就结束了，hyper-util 和 axum 都不再知道它的存在，停机既不通知它也不等它，进程退出时它被直接切断。
axum 的连接任务是单独 spawn 的，abort serve 任务不会停掉连接，只有运行时关闭才会。
修法：从发信号时开始计时的超时，超时后退出进程；长响应用 take_until 听停机信号；升级任务自己接收停机信号、自己被计数等待。

优雅停机一直不结束

复现一：GracefulShutdown

为什么：一个做不完，一个早就走了

复现二：axum 的 with_graceful_shutdown

修法

动脑筋答案

小结

💬 关联讨论

登录