feat: configure application transports and unify graceful shutdown

transport/native/HTTP2.md

@@ -42,4 +42,4 @@ HTTP/2 版本需要在 native `Stats` 之外至少记录：活动 stream、被
 - 代理联调：Nginx 的 TLS/h2 回源、Keep-Alive、超时/取消及 GOAWAY 排空已由 `verifyHTTP2Proxy.py` 验证。旧实例处理中的 POST 能正常完成，新请求转入备用实例，旧实例连接与请求归零。此测试使用单进程 Nginx，未验证 Nginx master/worker 重载或 Cloudflare 实网部署。`verifyProxy.py` 继续承担 native HTTP/1.1 TLS 回源验收。
 - 长测：Linux amd64 至少 30 分钟 TLS、多连接、多 stream、慢客户端混合场景；要求 0 协议错误、0 请求串线，连接、stream、goroutine 和 FD 在停止后回收。单连接场景与 8 连接慢上传/慢下载混合场景均已通过，原始日志与采样分别保存在两个报告中。
 
-这些实验结果已经覆盖到 Nginx HTTP/2 回源联调。下一步是统一启动配置、native 的 HTTP/1.1 与 HTTP/2 协议接入回归，以及业务环境试用，不能仅通过添加 `h2` 到 native 的 TLS ALPN 就宣告接入完成。当前继续保持显式选择实验适配器；默认 native 路径仍为 HTTP/1.1，需要成熟 HTTP/2 入口时可使用 `transport/nethttp` 或代理终止后回源 HTTP/1.1。
+统一启动配置已接入 `omni.New`：`GO_OMNI_TRANSPORT=http2` 加证书和私钥可显式选择实验适配器，业务路由不需要改变；native 仍只协商 HTTP/1.1。参考业务增加了三种适配器的真实网络验证与排空/超时回归。下一步是业务环境试用、native 的 HTTP/1.1 与 HTTP/2 多协议监听器设计和接入回归，不能仅通过添加 `h2` 到 native 的 TLS ALPN 就宣告接入完成。需要成熟 HTTP/2 入口时可使用 `transport/nethttp` 或代理终止后回源 HTTP/1.1。

transport/native/server.go

@@ -46,6 +46,7 @@ type Server struct {
 	ReadTimeout           time.Duration
 	WriteTimeout          time.Duration
 	IdleTimeout           time.Duration
+	ShutdownTimeout       time.Duration
 	ReadBufferSize        int
 	MaxHeaderBytes        int
 	MaxBodyBytes          int64
@@ -252,15 +253,14 @@ func (s *Server) Serve(ctx context.Context, listener net.Listener) error {
 	s.mu.Unlock()
 	defer cancel()
 
-	shutdownDone := make(chan struct{})
+	stopWatcher, watcherDone := make(chan struct{}), make(chan struct{})
 	go func() {
+		defer close(watcherDone)
 		select {
 		case <-ctx.Done():
-			timeout := s.shutdownTimeout()
-			shutdownCtx, cancel := context.WithTimeout(context.Background(), timeout)
-			_ = s.Shutdown(shutdownCtx)
-			cancel()
-		case <-shutdownDone:
+			// Wake Accept; the serving goroutine owns draining and its error.
+			_ = listener.Close()
+		case <-stopWatcher:
 		}
 	}()
 
@@ -307,20 +307,8 @@ func (s *Server) Serve(ctx context.Context, listener net.Listener) error {
 		go s.serveConnection(baseCtx, state)
 	}
 
-	close(shutdownDone)
-	s.mu.Lock()
-	if s.listener == listener {
-		s.listener = nil
-	}
-	s.mu.Unlock()
-	if ctx.Err() != nil || s.isClosing() {
-		timeout := s.shutdownTimeout()
-		shutdownCtx, cancel := context.WithTimeout(context.Background(), timeout)
-		shutdownErr := s.Shutdown(shutdownCtx)
-		cancel()
-		return errors.Join(shutdownErr, serveErr)
-	}
-	_ = listener.Close()
+	close(stopWatcher)
+	<-watcherDone
 	timeout := s.shutdownTimeout()
 	shutdownCtx, cancel := context.WithTimeout(context.Background(), timeout)
 	shutdownErr := s.Shutdown(shutdownCtx)
@@ -328,17 +316,15 @@ func (s *Server) Serve(ctx context.Context, listener net.Listener) error {
 	return errors.Join(shutdownErr, serveErr)
 }
 
-// Shutdown stops accepting requests, cancels active request contexts and waits
-// for connections to finish. On timeout, active sockets are closed.
+// Shutdown stops accepting requests and waits for active handlers to finish.
+// On timeout, request contexts are canceled and active sockets are closed.
 func (s *Server) Shutdown(ctx context.Context) error {
 	if ctx == nil {
 		ctx = context.Background()
 	}
 	s.mu.Lock()
 	s.closing = true
-	if s.cancel != nil {
-		s.cancel()
-	}
+	cancelRequests := s.cancel
 	listener := s.listener
 	states := make([]*connection, 0, len(s.conns))
 	for state := range s.conns {
@@ -365,6 +351,9 @@ func (s *Server) Shutdown(ctx context.Context) error {
 	case <-done:
 		return nil
 	case <-ctx.Done():
+		if cancelRequests != nil {
+			cancelRequests()
+		}
 		for _, state := range states {
 			state.close()
 		}
@@ -374,6 +363,9 @@ func (s *Server) Shutdown(ctx context.Context) error {
 }
 
 func (s *Server) shutdownTimeout() time.Duration {
+	if s.ShutdownTimeout > 0 {
+		return s.ShutdownTimeout
+	}
 	if value := s.WriteTimeout; value > 0 {
 		return value
 	}
@@ -432,9 +424,15 @@ func (s *Server) serveConnection(baseCtx context.Context, state *connection) {
 		}
 		_ = state.conn.SetReadDeadline(time.Time{})
 		firstRequest = false
+		s.mu.Lock()
+		if s.closing || state.closed.Load() {
+			s.mu.Unlock()
+			return
+		}
+		state.active.Store(true)
+		s.mu.Unlock()
 		requestContext, cancel := context.WithCancel(baseCtx)
 		request = request.WithContext(requestContext)
-		state.active.Store(true)
 		s.activeRequests.Add(1)
 		writer := s.writerPool.Get().(*responseWriter)
 		writer.reset(request.Method)

transport/native/server_test.go

@@ -504,12 +504,13 @@ func (b *closeBody) Close() error {
 	return errors.Join(err, b.conn.Close())
 }
 
-func TestServerShutdownCancelsActiveHandler(t *testing.T) {
-	started := make(chan struct{})
+func TestServerShutdownTimeoutCancelsActiveHandler(t *testing.T) {
+	started, canceled := make(chan struct{}), make(chan struct{})
 	application := omni.NewWithRouter(netrouter.New())
 	application.GET("/wait", func(ctx *omni.Context) error {
 		close(started)
 		<-ctx.Context().Done()
+		close(canceled)
 		return ctx.Context().Err()
 	})
 	server, listener, done := startServer(t, application)
@@ -524,19 +525,12 @@ func TestServerShutdownCancelsActiveHandler(t *testing.T) {
 	case <-time.After(2 * time.Second):
 		t.Fatal("handler did not start")
 	}
-	shutdownContext, cancel := context.WithTimeout(context.Background(), time.Second)
+	shutdownContext, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
 	defer cancel()
-	if err := server.Shutdown(shutdownContext); err != nil {
-		t.Fatal(err)
-	}
-	response, err := http.ReadResponse(bufio.NewReader(conn), &http.Request{Method: http.MethodGet})
-	if err != nil {
-		t.Fatal(err)
-	}
-	response.Body.Close()
-	if response.StatusCode != http.StatusInternalServerError {
-		t.Fatalf("expected canceled handler to become 500, got %d", response.StatusCode)
+	if err := server.Shutdown(shutdownContext); !errors.Is(err, context.DeadlineExceeded) {
+		t.Fatalf("expected shutdown deadline, got %v", err)
 	}
+	<-canceled
 	conn.Close()
 	select {
 	case err := <-done: