从连接到分发

loop {
    let accepted = tokio::select! {
        accepted = listener.accept() => accepted,
        () = &mut shutdown => break,
    };
    let (stream, remote_addr) = accepted?;
    let io = TokioIo::new(stream);
    let service = service.for_connection(remote_addr);
    // ...
    tokio::spawn(async move {
        // ...
        let builder = auto::Builder::new(TokioExecutor::new());
        let mut connection = pin!(builder.serve_connection_with_upgrades(io, service));
        // ...
    });
}

(ulimit -n 24; PORT=38104 exec ./dispatch)

connect 16 failed: [Errno 61] Connection refused
opened 16
after: failed [Errno 61] Connection refused

thread 'main' (2120768) panicked at src/bin/dispatch.rs:21:32:
called `Result::unwrap()` on an `Err` value: Os { code: 24, kind: TooManyOpenFiles, message: "Too many open files" }

async fn handle_accept_error(e: io::Error) {
    if is_connection_error(&e) {
        return;
    }
    // ...
    error!("accept error: {e}");
    tokio::time::sleep(Duration::from_secs(1)).await;
}

if let Err(_error) = result {
    // TODO: Surface real connection errors without the noise. Most
    // are benign (e.g. the client aborting an in-flight shard
    // request), so for now they are dropped. See how axum demotes
    // these to a `trace`-level log.
}

s.sendall(b"GET /api/items HTTP/1.1\r\nHost: x\r\n")

after 40.0s: still open, nothing received

// Each channel signals by closing, not by sending: connection tasks hold
// receiver clones, and dropping the sender resolves their `changed()`.
//
// Announces the shutdown; tasks switch into graceful shutdown.
let (drain_tx, drain_rx) = watch::channel(());
// Announces the end of the grace period; tasks drop their connection.
let (cutoff_tx, cutoff_rx) = watch::channel(());
// Tracks live tasks the other way around: each holds a receiver clone,
// and `done_tx.closed()` resolves once the last one is dropped.
let (done_tx, done_rx) = watch::channel(());

let result = tokio::select! {
    result = connection.as_mut() => result,
    _ = drain_rx.changed() => {
        // Finish the in-flight request, then close: HTTP/1 stops
        // keep-alive, HTTP/2 sends GOAWAY.
        connection.as_mut().graceful_shutdown();
        tokio::select! {
            result = connection.as_mut() => result,
            // The grace period ended; drop the connection as is.
            _ = cutoff_rx.changed() => return,
        }
    }
};

drop(listener);
// ...
drop(drain_rx);
drop(drain_tx);
drop(done_rx);
tokio::select! {
    () = done_tx.closed() => {}
    () = tokio::time::sleep(service.shutdown_timeout) => {}
}
// ...
drop(cutoff_rx);
drop(cutoff_tx);
done_tx.closed().await;

let listener = TcpListener::bind("127.0.0.1:38103").await?;
let service = RouterService::new(module_router!().build()).shutdown_timeout(Duration::from_secs(2));
let signal = async {
  STOP.notified().await;
  println!("[{:>5}ms] shutdown signal", t());
};
let result = topcoat::serve_until(listener, service, signal).await;
println!("[{:>5}ms] serve_until returned {result:?}", t());

-- new connection after stop:
curl exit=7
rebind ok while draining
[  507ms] slow1 start
[  507ms] slow5 start
[  724ms] shutdown signal
[ 1510ms] slow1 done
[ 2727ms] slow5 future dropped
[ 2727ms] serve_until returned Ok(())
-- slow1:
[ 1007ms] read 147 bytes:
HTTP/1.1 200 OK
content-type: text/plain; charset=utf-8
connection: close
content-length: 11

slow1 done

[ 1007ms] <EOF>
-- slow5:
[ 2221ms] <EOF>

impl Service<Request<Incoming>> for RouterService {
    type Response = Response;
    type Error = Infallible;
    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;

    fn call(&self, mut request: Request<Incoming>) -> Self::Future {
        let router = self.router.clone();
        if let Some(addr) = self.remote_addr {
            request.extensions_mut().insert(RemoteAddr(addr));
        }
        Box::pin(async move { Ok(router.handle(request.map(Body::new)).await) })
    }
}

let mut future = pin!(self.handle_inner(request, RewriteChain::carrying(values)));

poll_fn(|cx| {
    // The whole request and its context are discarded after a panic,
    // so no potentially inconsistent request-local state is reused.
    match catch_unwind(AssertUnwindSafe(|| future.as_mut().poll(cx))) {
        Ok(Poll::Ready(response)) => Poll::Ready(response),
        Ok(Poll::Pending) => Poll::Pending,
        Err(_) => Poll::Ready(internal_server_response()),
    }
})
.await

### http://127.0.0.1:38102/boom
HTTP/1.1 500 Internal Server Error
content-length: 21

internal server error

### http://127.0.0.1:38102/api/items
HTTP/1.1 200 OK
content-type: text/plain; charset=utf-8
content-length: 6

items

// Resolve the client's address once from the original request,
// before any rewrite can replace its headers.
let client_ip = ClientIp(inner.trusted_proxies.resolve(&request));
let (mut parts, mut body) = request.into_parts();
let original = OriginalParts(Arc::new(parts.clone()));

let (cx, result) = loop {
    // Rewrites start with a fresh context. Keep only the values
    // passed by the caller or a rewrite, and discard any other
    // request state. Add the router's values last so they win.
    let cx = Cx::new(Arc::clone(&inner.app_context))
        .with_many(chain.context().clone())
        .with_many((
            Arc::clone(&self.inner),
            ResponseHeaders::new(),
            RawPathParams::default(),
            original.clone(),
            client_ip,
            parts,
        ));
    // ...
    let (cx, result) = self.dispatch(cx, body).await;
    // ...
};

let rewrite = match result {
    Ok(response) => break (cx, Ok(response)),
    Err(error) => match error.downcast::<RewriteError>() {
        Ok(rewrite) => rewrite,
        Err(failed) => break (cx, Err(failed.into_error())),
    },
};
(parts, body) = match chain.follow(crate::request::parts(&cx), rewrite) {
    Ok(request) => request.into_parts(),
    Err(error) => break (cx, Err(error)),
};

  pathless  > GET /old
  root      > /old
  handler old: rewrite to /api/items (/old)
  root      < Err(rewrite to /api/items)
  pathless  < Err(rewrite to /api/items)
  pathless  > GET /api/items
  root      > /api/items
  api       > /api/items
  handler list_items (GET)
  api       < 200 OK
  root      < 200 OK
  pathless  < 200 OK
  pathless  > GET /loop-a
  root      > /loop-a
  root      < Err(rewrite to /loop-b)
  pathless  < Err(rewrite to /loop-b)
  pathless  > GET /loop-b
  root      > /loop-b
  root      < Err(rewrite to /loop-a)
  pathless  < Err(rewrite to /loop-a)

### -I http://127.0.0.1:38102/api/items
HTTP/1.1 200 OK
content-type: text/plain; charset=utf-8
content-length: 6

### -X DELETE http://127.0.0.1:38102/api/items
HTTP/1.1 405 Method Not Allowed
allow: GET, POST, HEAD
content-length: 0

### http://127.0.0.1:38102/nope
HTTP/1.1 404 Not Found
content-type: text/plain; charset=utf-8
content-length: 9

not found

### -X POST -H Sec-Fetch-Site: cross-site http://127.0.0.1:38102/api/items
HTTP/1.1 403 Forbidden
content-type: text/plain; charset=utf-8
content-length: 9

forbidden

  pathless  > HEAD /api/items
  root      > /api/items
  api       > /api/items
  handler list_items (HEAD)
  api       < 200 OK
  root      < 200 OK
  pathless  < 200 OK
  pathless  > DELETE /api/items
  pathless  < Err(method not allowed)
  pathless  > GET /nope
  pathless  < Err(not found)