diff --git a/README.md b/README.md index b00fe17..a4cd342 100644 --- a/README.md +++ b/README.md @@ -175,6 +175,33 @@ application.OnShutdown(logger.Close) `X-Forwarded-For`,默认只记录 socket 对端地址。数据库连接仍由 `database.ConfigFromEnv` 从 `GO_OMNI_DB_*` 加载。 +网络层也由同一份配置选择,Handler / Service / Repository 不需要导入具体适配器: + +| 配置 | 默认值 / 用途 | +| --- | --- | +| `GO_OMNI_TRANSPORT` | `nethttp`;可选 `native`、`http2` | +| `GO_OMNI_TLS_CERT_FILE` | TLS 证书 PEM 文件路径 | +| `GO_OMNI_TLS_KEY_FILE` | TLS 私钥 PEM 文件路径,必须与证书一起配置 | +| `GO_OMNI_SHUTDOWN_TIMEOUT` | `10s`,三种适配器共用的请求排空时间 | + +例如运行自研 HTTP/1.1,或显式试用独立 HTTP/2 适配器: + +```sh +GO_OMNI_TRANSPORT=native go run ./examples/reference +GO_OMNI_TRANSPORT=http2 GO_OMNI_HTTP_ADDR=127.0.0.1:8443 \ + GO_OMNI_TLS_CERT_FILE=./server.crt GO_OMNI_TLS_KEY_FILE=./server.key \ + go run ./examples/reference +``` + +`http2` 必须配置 TLS;native TLS 只协商 `http/1.1`,nethttp TLS 支持标准库的 HTTP/1.1 和 HTTP/2。 +未知适配器、缺少配对证书、无法加载证书时,`Run` / `RunContext` 在监听前返回错误并执行资源清理,不会退回其他协议。 +HTTP/2 仍为实验适配器,尚未合并到 native 多协议监听器。 + +代码配置可写成 `omni.New(omni.WithTransport(omni.TransportNative), omni.WithTLS(certFile, keyFile))`。 +`ServeContext(ctx, listener)` 使用相同配置并接管已有 listener;失败或退出时也会关闭它。 +`RunWith(ctx, runner)` 保留自定义网络层入口,由 runner 自己管理 TLS 和网络参数;连接/请求上限等适配器专属选项仍通过这个入口设置。 +`NewServer` 仅创建标准库明文服务器,配置选择应用于 `Run`、`RunContext` 和 `ServeContext`。 + `App` 同时实现标准 `http.Handler`。`Run` 监听 `SIGINT` / `SIGTERM`;`RunContext` 则由调用方的 context 控制退出,生成项目已使用 `signal.NotifyContext` 接好信号。 退出时先停止接收新请求,等待活跃请求完成,再执行资源清理。超过 `ShutdownTimeout` 时强制关闭连接、取消请求 context,并返回超时错误。业务代码应响应 context 取消;框架无法强制结束忽略取消的业务 goroutine。数据库连接池由 `database.Open` 创建,在服务退出时调用 `Close`。 diff --git a/ROADMAP.md b/ROADMAP.md index 1523f14..d135f5a 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -58,7 +58,9 @@ - 已补连接级窗口耗尽时的暂停、同连接 204/跨连接隔离、WINDOW_UPDATE 恢复和连接复用;已补零窗口写阻塞在关闭超时后的解除阻塞、请求取消与回收。Linux amd64 的 HTTP/2 全包 race、vet、build 已通过,见 [边界回归报告](performance/http2FlowControlLinux.json)。 - 新增 HTTP/2 零 stream 窗口时的慢响应隔离、WINDOW_UPDATE 恢复、RST_STREAM 取消与请求槽位回收测试;补充 TLS 半开握手占满连接上限时的及时拒绝、超时回收及正常请求恢复测试。 - 已完成 Nginx 1.30.5 → HTTP/2 的 TLS 回源联调:普通与 race 模式各 3 轮,共 18 个子场景通过;覆盖两段证书校验、32 次 POST 复用一条回源连接、不可信 CA/主机名拒绝、504 超时、客户端取消、GOAWAY 后旧 POST 完成及 20 个新 POST 转入新实例,无重复业务执行。见 [验收说明与报告](transport/http2/PROXY.md)。 -- `scripts/verifyHTTP2Proxy.py` 顺序运行受限编译容器和禁用外部网络的测试容器;总时限、SIGTERM、SIGHUP 清理已在 Linux 实测,原有 7 个业务容器仍运行。HTTP/2 下一步是统一启动配置、native 多协议接入回归与业务试用;当前结果不覆盖 Nginx master/worker 重载或 Cloudflare 实网部署。 +- `scripts/verifyHTTP2Proxy.py` 顺序运行受限编译容器和禁用外部网络的测试容器;总时限、SIGTERM、SIGHUP 清理已在 Linux 实测,原有 7 个业务容器仍运行。当前结果不覆盖 Nginx master/worker 重载或 Cloudflare 实网部署。 +- 已接入 `omni.New` 的统一 transport / TLS / shutdownTimeout 配置:默认 nethttp,可显式选择 native 或实验 HTTP/2;证书在监听前加载,HTTP/2 配置必须启用 TLS。`RunWith` 继续支持自定义网络层,业务不依赖具体适配器。 +- 参考业务已增加五种真实 TCP/TLS 组合的 CRUD、认证权限、错误响应、requestId 日志与协议协商测试;三种适配器共用排空和超时退出测试。修复 native 退出时过早取消在途请求,并使关闭超时错误返回调用方。下一步仍为业务环境试用、native 多协议监听器设计与接入回归,HTTP/2 保持实验状态。 - 已用 16 个并发 worker 完成 3,200 次建连/断连 race 验收;MaxConnections 满载拒绝、槽位释放和 TLS 连接上限均已覆盖。 - 新增 `scripts/verifyLinux.py`,固定在 Go 1.25 Linux 容器内运行 race、vet、build 和可选 TLS/非 TLS perfcheck;perfcheck 记录堆、goroutine、连接和 Linux 文件描述符。 - Linux/arm64 已完成 30 秒、128 并发、512 空闲连接的 TLS 与非 TLS 持续回归,报告见 `performance/linuxSoak.json` 和 `performance/linuxSoakTLS.json`,均为 0 错误且连接、goroutine、FD 回收正常。 diff --git a/app/app.go b/app/app.go index 04b6881..f3e99d4 100644 --- a/app/app.go +++ b/app/app.go @@ -31,7 +31,7 @@ type App struct { } // Runner is the small lifecycle boundary used by an HTTP server -// implementation. The default App.RunContext uses net/http; a custom runner +// implementation. App.RunContext selects a configured built-in server; a custom runner // can own the listener, protocol parser and connection lifecycle. type Runner interface { Run(context.Context, http.Handler, string) error @@ -133,7 +133,8 @@ func (a *App) Run(address ...string) error { return a.RunContext(ctx, address...) } -// NewServer creates a standard HTTP server around the application. +// NewServer creates a plain standard HTTP server around the application. +// Use RunContext or ServeContext to apply the configured transport and TLS. func (a *App) NewServer(address string) *http.Server { if address == "" { address = a.config.Address @@ -152,23 +153,21 @@ func (a *App) RunContext(ctx context.Context, address ...string) error { if ctx.Err() != nil { return a.Close() } - listenOn := a.config.Address - if listenOn == "" { - listenOn = ":8080" - } - if len(address) > 0 && address[0] != "" { - listenOn = address[0] - } - server := a.NewServer(listenOn) - listener, err := net.Listen("tcp", server.Addr) + serve, err := a.prepareTransport() if err != nil { return errors.Join(err, a.Close()) } - return a.serveContext(ctx, server, listener) + listener, err := net.Listen("tcp", a.listenAddress(address...)) + if err != nil { + return errors.Join(err, a.Close()) + } + defer listener.Close() + return errors.Join(serve(ctx, listener), a.Close()) } // RunWith starts a custom server runner while keeping App-owned cleanup in the -// same lifecycle. The runner receives the transport-neutral App handler. +// same lifecycle. The runner receives the transport-neutral App handler and +// owns transport/TLS settings; built-in transport selection is bypassed. func (a *App) RunWith(ctx context.Context, runner Runner, address ...string) error { if runner == nil { return errors.Join(errors.New("go-omni: nil runner"), a.Close()) @@ -176,6 +175,13 @@ func (a *App) RunWith(ctx context.Context, runner Runner, address ...string) err if ctx == nil { ctx = context.Background() } + if ctx.Err() != nil { + return a.Close() + } + return errors.Join(runner.Run(ctx, a, a.listenAddress(address...)), a.Close()) +} + +func (a *App) listenAddress(address ...string) string { listenOn := a.config.Address if listenOn == "" { listenOn = ":8080" @@ -183,12 +189,18 @@ func (a *App) RunWith(ctx context.Context, runner Runner, address ...string) err if len(address) > 0 && address[0] != "" { listenOn = address[0] } - return errors.Join(runner.Run(ctx, a, listenOn), a.Close()) + return listenOn } func (a *App) serveContext(ctx context.Context, server *http.Server, listener net.Listener) error { serveErr := make(chan error, 1) - go func() { serveErr <- server.Serve(listener) }() + go func() { + if server.TLSConfig != nil { + serveErr <- server.ServeTLS(listener, "", "") + } else { + serveErr <- server.Serve(listener) + } + }() select { case err := <-serveErr: diff --git a/app/transport.go b/app/transport.go new file mode 100644 index 0000000..f6e6a64 --- /dev/null +++ b/app/transport.go @@ -0,0 +1,79 @@ +package app + +import ( + "context" + "crypto/tls" + "errors" + "fmt" + "net" + "strings" + "time" + + "git.yuntuo.hk/k2safe/go-omni/config" + h2transport "git.yuntuo.hk/k2safe/go-omni/transport/http2" + "git.yuntuo.hk/k2safe/go-omni/transport/native" +) + +// ServeContext serves a caller-created listener using the configured transport +// and TLS. App takes ownership of the listener, including on startup failure, +// and runs resource cleanup after the server has stopped. +func (a *App) ServeContext(ctx context.Context, listener net.Listener) error { + if listener == nil { + return errors.Join(errors.New("go-omni: nil listener"), a.Close()) + } + defer listener.Close() + if ctx == nil { + ctx = context.Background() + } + if ctx.Err() != nil { + _ = listener.Close() + return a.Close() + } + serve, err := a.prepareTransport() + if err != nil { + _ = listener.Close() + return errors.Join(err, a.Close()) + } + return errors.Join(serve(ctx, listener), a.Close()) +} + +func (a *App) prepareTransport() (func(context.Context, net.Listener) error, error) { + if err := a.config.ValidateTransport(); err != nil { + return nil, err + } + shutdownTimeout := a.config.ShutdownTimeout + if shutdownTimeout <= 0 { + shutdownTimeout = 10 * time.Second + } + var tlsConfig *tls.Config + if strings.TrimSpace(a.config.TLS.CertFile) != "" { + certificate, err := tls.LoadX509KeyPair(a.config.TLS.CertFile, a.config.TLS.KeyFile) + if err != nil { + return nil, fmt.Errorf("go-omni: load TLS certificate: %w", err) + } + tlsConfig = &tls.Config{Certificates: []tls.Certificate{certificate}, MinVersion: tls.VersionTLS12} + } + switch a.config.Transport { + case config.TransportNative: + server := native.New(a) + server.ShutdownTimeout = shutdownTimeout + if tlsConfig != nil { + return func(ctx context.Context, listener net.Listener) error { + return server.ServeTLS(ctx, listener, tlsConfig) + }, nil + } + return server.Serve, nil + case config.TransportHTTP2: + server := h2transport.New(a) + server.ShutdownTimeout = shutdownTimeout + return func(ctx context.Context, listener net.Listener) error { + return server.ServeTLS(ctx, listener, tlsConfig) + }, nil + default: + server := a.NewServer("") + server.TLSConfig = tlsConfig + return func(ctx context.Context, listener net.Listener) error { + return a.serveContext(ctx, server, listener) + }, nil + } +} diff --git a/app/transport_test.go b/app/transport_test.go new file mode 100644 index 0000000..c4d53fd --- /dev/null +++ b/app/transport_test.go @@ -0,0 +1,266 @@ +package app + +import ( + "context" + "errors" + "io" + "net" + "net/http" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "git.yuntuo.hk/k2safe/go-omni/config" + "git.yuntuo.hk/k2safe/go-omni/httpx" + "git.yuntuo.hk/k2safe/go-omni/internal/testutil" +) + +func TestConfiguredTransportLifecycle(t *testing.T) { + for _, transport := range []string{config.TransportNetHTTP, config.TransportNative, config.TransportHTTP2} { + for _, timeout := range []bool{false, true} { + name := transport + "/drain" + if timeout { + name = transport + "/timeout" + } + t.Run(name, func(t *testing.T) { + cert, key, tlsClient := testutil.CertificateFiles(t) + limit := 3 * time.Second + if timeout { + limit = 80 * time.Millisecond + } + application := New(config.WithTransport(transport), config.WithTLS(cert, key), config.WithShutdownTimeout(limit)) + started, release, canceled := make(chan struct{}), make(chan struct{}), make(chan struct{}) + var once sync.Once + unblock := func() { once.Do(func() { close(release) }) } + defer unblock() + var cleaned atomic.Int32 + application.OnShutdown(func() error { cleaned.Add(1); return nil }) + application.GET("/slow", func(ctx *httpx.Context) error { + close(started) + select { + case <-ctx.Context().Done(): + close(canceled) + return ctx.Context().Err() + case <-release: + return ctx.OK("finished") + } + }) + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + wrapped := &closingListener{Listener: listener, closed: make(chan struct{})} + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + defer listener.Close() + done := make(chan error, 1) + go func() { done <- application.ServeContext(ctx, wrapped) }() + client := &http.Client{Timeout: 5 * time.Second, Transport: &http.Transport{TLSClientConfig: tlsClient, ForceAttemptHTTP2: true}} + defer client.CloseIdleConnections() + requestDone := make(chan error, 1) + go func() { + response, err := client.Get("https://" + listener.Addr().String() + "/slow") + if err == nil { + defer response.Body.Close() + body, readErr := io.ReadAll(response.Body) + err = readErr + if response.StatusCode != 200 || !strings.Contains(string(body), "finished") { + err = errors.New("drained response was lost") + } + } + requestDone <- err + }() + await(t, started) + began := time.Now() + cancel() + await(t, wrapped.closed) + if !timeout { + select { + case <-canceled: + t.Fatal("active request canceled before drain") + default: + } + if cleaned.Load() != 0 { + t.Fatal("cleanup preceded active request") + } + unblock() + } + select { + case err := <-done: + if timeout { + if !errors.Is(err, context.DeadlineExceeded) || time.Since(began) < limit { + t.Fatalf("timeout ignored: %v after %v", err, time.Since(began)) + } + await(t, canceled) + } else if err != nil { + t.Fatal(err) + } + case <-time.After(5 * time.Second): + t.Fatal("server failed to stop") + } + select { + case err := <-requestDone: + if !timeout && err != nil { + t.Fatal(err) + } + case <-time.After(5 * time.Second): + t.Fatal("client failed to finish") + } + _ = application.Close() + if cleaned.Load() != 1 { + t.Fatalf("cleanup count %d", cleaned.Load()) + } + }) + } + } +} + +func TestConfiguredStartupFailureClosesResources(t *testing.T) { + occupied, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer occupied.Close() + for _, test := range []struct { + name, errorText string + options []config.Option + }{ + {"unknown", "transport", []config.Option{config.WithTransport("unknown")}}, + {"http2WithoutTLS", "tls", []config.Option{config.WithTransport("http2")}}, + {"partialTLS", "tls", []config.Option{config.WithTLS("missing.crt", "")}}, + {"nethttpCertificate", "load TLS certificate", []config.Option{config.WithTLS("missing.crt", "missing.key")}}, + {"nativeCertificate", "load TLS certificate", []config.Option{config.WithTransport("native"), config.WithTLS("missing.crt", "missing.key")}}, + {"http2Certificate", "load TLS certificate", []config.Option{config.WithTransport("http2"), config.WithTLS("missing.crt", "missing.key")}}, + } { + t.Run(test.name, func(t *testing.T) { + application := New(test.options...) + cleanupError := errors.New("cleanup failure") + count := 0 + application.OnShutdown(func() error { count++; return cleanupError }) + // The occupied address must not mask configuration/certificate errors. + err := application.RunContext(context.Background(), occupied.Addr().String()) + if err == nil || !strings.Contains(err.Error(), test.errorText) || !errors.Is(err, cleanupError) { + t.Fatalf("startup: %v", err) + } + _ = application.Close() + if count != 1 { + t.Fatalf("cleanup count %d", count) + } + }) + } + for _, canceled := range []bool{false, true} { + application := New(config.WithTransport("unknown")) + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(context.Background()) + if canceled { + cancel() + } + err = application.ServeContext(ctx, listener) + cancel() + if canceled && err != nil { + t.Fatal(err) + } + if !canceled && err == nil { + t.Fatal("invalid transport accepted") + } + if _, err := listener.Accept(); !errors.Is(err, net.ErrClosed) { + t.Fatalf("listener leaked: %v", err) + } + } +} + +func TestRunContextConfiguredTLS(t *testing.T) { + for _, transport := range []string{config.TransportNetHTTP, config.TransportNative, config.TransportHTTP2} { + t.Run(transport, func(t *testing.T) { + cert, key, tlsClient := testutil.CertificateFiles(t) + reserved, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + address := reserved.Addr().String() + reserved.Close() + settings := config.Default() + settings.Address, settings.Transport = address, transport + settings.TLS = config.TLSConfig{CertFile: cert, KeyFile: key} + application := New(config.WithConfig(settings)) + application.GET("/health", func(ctx *httpx.Context) error { return ctx.OK("ok") }) + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- application.RunContext(ctx) }() + defer func() { + cancel() + select { + case err := <-done: + if err != nil { + t.Error(err) + } + case <-time.After(5 * time.Second): + t.Error("RunContext failed to stop") + } + }() + client := &http.Client{Timeout: time.Second, Transport: &http.Transport{TLSClientConfig: tlsClient, ForceAttemptHTTP2: true}} + defer client.CloseIdleConnections() + deadline := time.Now().Add(3 * time.Second) + for { + response, err := client.Get("https://" + address + "/health") + if err == nil { + body, readErr := io.ReadAll(response.Body) + response.Body.Close() + if readErr != nil || response.StatusCode != 200 || !strings.Contains(string(body), `"data":"ok"`) { + t.Fatalf("response %s: %v", body, readErr) + } + want := "h2" + if transport == config.TransportNative { + want = "http/1.1" + } + if response.TLS.NegotiatedProtocol != want || len(response.TLS.VerifiedChains) == 0 { + t.Fatalf("TLS negotiation: %+v", response.TLS) + } + break + } + if time.Now().After(deadline) { + t.Fatal(err) + } + time.Sleep(10 * time.Millisecond) + } + }) + } +} + +type runnerFunc func(context.Context, http.Handler, string) error + +func (f runnerFunc) Run(ctx context.Context, handler http.Handler, address string) error { + return f(ctx, handler, address) +} + +func TestRunWithRemainsIndependentOfBuiltInTransport(t *testing.T) { + for _, canceled := range []bool{false, true} { + application := New(config.WithTransport("custom"), config.WithTLS("missing", "missing")) + ctx, cancel := context.WithCancel(context.Background()) + if canceled { + cancel() + } + calls, cleanups := 0, 0 + application.OnShutdown(func() error { cleanups++; return nil }) + err := application.RunWith(ctx, runnerFunc(func(_ context.Context, handler http.Handler, address string) error { + calls++ + if handler != application || address != "127.0.0.1:1234" { + t.Fatal("custom runner arguments changed") + } + return nil + }), "127.0.0.1:1234") + cancel() + wantCalls := 1 + if canceled { + wantCalls = 0 + } + if err != nil || calls != wantCalls || cleanups != 1 { + t.Fatalf("RunWith: %v calls=%d cleanups=%d", err, calls, cleanups) + } + } +} diff --git a/cmd/go-omni/scaffold/templates/README.md.tmpl b/cmd/go-omni/scaffold/templates/README.md.tmpl index a6fd672..f4e6a68 100644 --- a/cmd/go-omni/scaffold/templates/README.md.tmpl +++ b/cmd/go-omni/scaffold/templates/README.md.tmpl @@ -47,6 +47,11 @@ go build -o server ./cmd/server ``` `GO_OMNI_SHUTDOWN_TIMEOUT` 控制等待请求完成的时间,默认 `10s`。 + +使用支持统一网络配置的框架版本时,设置 `GO_OMNI_TRANSPORT=native` 即可切换为自研 HTTP/1.1; +默认 `nethttp`。`GO_OMNI_TLS_CERT_FILE` 和 `GO_OMNI_TLS_KEY_FILE` 配对启用 HTTPS。 +实验版 HTTP/2 使用 `GO_OMNI_TRANSPORT=http2`,必须同时配置 TLS 证书和私钥。 +业务 Handler / Service / Repository 不需要修改;三种适配器共用上述退出等待时间。 {{if eq .Database "mysql"}} `/health/ready` 已注册数据库探针:连接失败返回 `503`,数据库恢复后重新返回 `200`。 `/health/live` 始终只检查服务存活。`GO_OMNI_DB_PING_TIMEOUT` 限制初始化检查和数据库探针的等待时间。 diff --git a/config/config.go b/config/config.go index 78fb187..cb66d3f 100644 --- a/config/config.go +++ b/config/config.go @@ -10,19 +10,34 @@ import ( const defaultRequestIDHeader = "X-Request-ID" +const ( + TransportNetHTTP = "nethttp" + TransportNative = "native" + TransportHTTP2 = "http2" +) + +// TLSConfig enables HTTPS when both certificate paths are provided. +type TLSConfig struct { + CertFile string `json:"certFile"` + KeyFile string `json:"keyFile"` +} + // Config controls the default runtime behavior of an App. type Config struct { - Address string - RequestIDHeader string - EnableLogger bool - TrustedProxies []string - ShutdownTimeout time.Duration - Log logx.Config + Address string `json:"address"` + Transport string `json:"transport"` + TLS TLSConfig `json:"tls"` + RequestIDHeader string `json:"requestIdHeader"` + EnableLogger bool `json:"enableLogger"` + TrustedProxies []string `json:"trustedProxies"` + ShutdownTimeout time.Duration `json:"shutdownTimeout"` + Log logx.Config `json:"log"` } func Default() Config { return Config{ Address: ":8080", + Transport: TransportNetHTTP, RequestIDHeader: defaultRequestIDHeader, ShutdownTimeout: 10 * time.Second, Log: logx.Default(), @@ -43,6 +58,24 @@ func (c Config) Validate() error { if _, err := httpx.NewClientIPResolver(c.TrustedProxies...); err != nil { return &ConfigError{Field: "trusted_proxies", Message: err.Error()} } + return c.ValidateTransport() +} + +// ValidateTransport checks built-in transport selection without opening files +// or sockets. Empty Transport preserves the standard net/http default. +func (c Config) ValidateTransport() error { + switch c.Transport { + case "", TransportNetHTTP, TransportNative, TransportHTTP2: + default: + return &ConfigError{Field: "transport", Message: "must be nethttp, native or http2"} + } + cert, key := strings.TrimSpace(c.TLS.CertFile), strings.TrimSpace(c.TLS.KeyFile) + if (cert == "") != (key == "") { + return &ConfigError{Field: "tls", Message: "requires both certFile and keyFile"} + } + if c.Transport == TransportHTTP2 && cert == "" { + return &ConfigError{Field: "tls", Message: "is required for the experimental http2 transport"} + } return nil } @@ -57,6 +90,16 @@ func (e *ConfigError) Error() string { return "config " + e.Field + " " + e.Mess // Option changes an App's configuration. type Option func(*Config) +// WithTransport selects a built-in server. HTTP/2 is experimental and requires TLS. +func WithTransport(name string) Option { + return func(cfg *Config) { cfg.Transport = name } +} + +// WithTLS enables HTTPS. Certificates are loaded before Run opens its listener. +func WithTLS(certFile, keyFile string) Option { + return func(cfg *Config) { cfg.TLS = TLSConfig{CertFile: certFile, KeyFile: keyFile} } +} + // WithRequestIDHeader changes the header used to read and return request IDs. func WithRequestIDHeader(header string) Option { return func(cfg *Config) { diff --git a/config/config_test.go b/config/config_test.go index da2141b..0e1e834 100644 --- a/config/config_test.go +++ b/config/config_test.go @@ -10,6 +10,7 @@ func TestLoadFromEnvUsesDefaults(t *testing.T) { for _, key := range []string{ "GO_OMNI_HTTP_ADDR", "GO_OMNI_REQUEST_ID_HEADER", "GO_OMNI_ENABLE_LOGGER", "GO_OMNI_TRUSTED_PROXIES", + "GO_OMNI_TRANSPORT", "GO_OMNI_TLS_CERT_FILE", "GO_OMNI_TLS_KEY_FILE", "GO_OMNI_LOG_LEVEL", "GO_OMNI_LOG_FORMAT", "GO_OMNI_LOG_OUTPUT", "GO_OMNI_LOG_FILE", "GO_OMNI_SERVICE", "GO_OMNI_ENVIRONMENT", "GO_OMNI_LOG_ADD_SOURCE", "GO_OMNI_SHUTDOWN_TIMEOUT", @@ -21,7 +22,7 @@ func TestLoadFromEnvUsesDefaults(t *testing.T) { if err != nil { t.Fatal(err) } - if cfg.Address != ":8080" || cfg.RequestIDHeader != "X-Request-ID" || cfg.ShutdownTimeout != 10*time.Second { + if cfg.Address != ":8080" || cfg.Transport != TransportNetHTTP || cfg.TLS != (TLSConfig{}) || cfg.RequestIDHeader != "X-Request-ID" || cfg.ShutdownTimeout != 10*time.Second { t.Fatalf("unexpected defaults: %+v", cfg) } if cfg.EnableLogger || cfg.Log.Level != "info" || cfg.Log.Format != "json" || cfg.Log.Output != "stderr" { @@ -29,6 +30,44 @@ func TestLoadFromEnvUsesDefaults(t *testing.T) { } } +func TestTransportEnvironment(t *testing.T) { + for _, test := range []struct { + name, transport, cert, key, errorField string + }{ + {"default", "", "", "", ""}, + {"native", "native", "", "", ""}, + {"nativeTLS", "native", "server.crt", "server.key", ""}, + {"http2", "http2", "server.crt", "server.key", ""}, + {"unknown", "typo", "", "", "transport"}, + {"certOnly", "nethttp", "server.crt", "", "tls"}, + {"keyOnly", "native", "", "server.key", "tls"}, + {"http2WithoutTLS", "http2", "", "", "tls"}, + } { + t.Run(test.name, func(t *testing.T) { + t.Setenv("TRANSPORT_TEST_TRANSPORT", test.transport) + t.Setenv("TRANSPORT_TEST_TLS_CERT_FILE", test.cert) + t.Setenv("TRANSPORT_TEST_TLS_KEY_FILE", test.key) + cfg, err := LoadFromEnv("TRANSPORT_TEST_") + if test.errorField != "" { + if err == nil || !strings.Contains(err.Error(), test.errorField) { + t.Fatalf("unexpected error: %v", err) + } + return + } + if err != nil { + t.Fatal(err) + } + want := test.transport + if want == "" { + want = TransportNetHTTP + } + if cfg.Transport != want || cfg.TLS.CertFile != test.cert || cfg.TLS.KeyFile != test.key { + t.Fatalf("settings: %+v", cfg) + } + }) + } +} + func TestLoadFromEnvOverridesSettings(t *testing.T) { t.Setenv("GO_OMNI_HTTP_ADDR", "127.0.0.1:9090") t.Setenv("GO_OMNI_REQUEST_ID_HEADER", "X-Correlation-ID") diff --git a/config/env.go b/config/env.go index 7ca2aa3..fa200aa 100644 --- a/config/env.go +++ b/config/env.go @@ -19,6 +19,9 @@ func LoadFromEnv(prefix string) (Config, error) { } cfg := Default() cfg.Address = envString(prefix+"HTTP_ADDR", cfg.Address) + cfg.Transport = envString(prefix+"TRANSPORT", cfg.Transport) + cfg.TLS.CertFile = envString(prefix+"TLS_CERT_FILE", "") + cfg.TLS.KeyFile = envString(prefix+"TLS_KEY_FILE", "") cfg.RequestIDHeader = envString(prefix+"REQUEST_ID_HEADER", cfg.RequestIDHeader) if value := strings.TrimSpace(os.Getenv(prefix + "TRUSTED_PROXIES")); value != "" { cfg.TrustedProxies = strings.Split(value, ",") diff --git a/examples/reference/main.go b/examples/reference/main.go index aa54170..c386a1d 100644 --- a/examples/reference/main.go +++ b/examples/reference/main.go @@ -2,6 +2,7 @@ package main import ( "context" + "log" "os" "os/signal" "strings" @@ -41,28 +42,52 @@ func registerRoutes(application *omni.App, users service.UserRepository, logger } func main() { - logger, err := logx.New(logx.Config{ - Service: "reference-api", - Environment: "local", - Format: "json", - Writer: os.Stdout, - }) - if err != nil { - os.Exit(1) - } - defer logger.Close() ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() + if err := runReference(ctx); err != nil { + log.Fatal(err) + } +} + +func referenceConfig() (omni.Config, error) { + settings, err := omni.LoadConfig("") + if err != nil { + return settings, err + } + // Preserve the example's loopback default and legacy address variable. + if strings.TrimSpace(os.Getenv("GO_OMNI_HTTP_ADDR")) == "" { + settings.Address = strings.TrimSpace(os.Getenv("GO_OMNI_ADDR")) + if settings.Address == "" { + settings.Address = "127.0.0.1:8080" + } + } + if settings.Log.Service == "" { + settings.Log.Service = "reference-api" + } + if settings.Log.Environment == "" { + settings.Log.Environment = "local" + } + if strings.TrimSpace(os.Getenv("GO_OMNI_LOG_OUTPUT")) == "" { + settings.Log.Output = "stdout" + } + return settings, nil +} + +func runReference(ctx context.Context) error { + settings, err := referenceConfig() + if err != nil { + return err + } + logger, err := logx.New(settings.Log) + if err != nil { + return err + } + defer logger.Close() + logx.SetDefault(logger) users, closeStorage, storage, err := openUsers(ctx, logger) if err != nil { - logger.Error("storage initialization failed", "error", err) - os.Exit(1) + return err } - defer func() { - if err := closeStorage(); err != nil { - logger.Error("storage close failed", "error", err) - } - }() devToken := strings.TrimSpace(os.Getenv("GO_OMNI_DEV_TOKEN")) if devToken == "" { devToken = "dev-token" @@ -70,16 +95,11 @@ func main() { authenticator := auth.StaticAuthenticator{ devToken: {UserID: "dev-user", Roles: []string{"admin"}}, } - application := newProtectedApp(users, logger, authenticator) - address := os.Getenv("GO_OMNI_ADDR") - if address == "" { - address = "127.0.0.1:8080" - } - logger.Info("starting reference API", "address", address, "storage", storage) - if err := application.RunContext(ctx, address); err != nil { - logger.Error("server stopped", "error", err) - os.Exit(1) - } + application := newProtectedApp(users, logger, authenticator, omni.WithConfig(settings)) + application.OnShutdown(logger.Close) + application.OnShutdown(closeStorage) + logger.Info("starting reference API", "address", settings.Address, "transport", settings.Transport, "tls", settings.TLS.CertFile != "", "storage", storage) + return application.RunContext(ctx) } func openUsers(ctx context.Context, logger *logx.Logger) (service.UserRepository, func() error, string, error) { diff --git a/examples/reference/main_test.go b/examples/reference/main_test.go index ebcbe52..d5e3287 100644 --- a/examples/reference/main_test.go +++ b/examples/reference/main_test.go @@ -118,7 +118,7 @@ func TestUserCRUD(t *testing.T) { } } -func checkUserCRUD(t *testing.T, app *omni.App) { +func checkUserCRUD(t *testing.T, app http.Handler) { t.Helper() health := dataAs[map[string]string](t, request(t, app, "GET", "/health", "", 200, "OK")) if health["status"] != "ok" { diff --git a/examples/reference/transport_test.go b/examples/reference/transport_test.go new file mode 100644 index 0000000..5b17889 --- /dev/null +++ b/examples/reference/transport_test.go @@ -0,0 +1,182 @@ +package main + +import ( + "bytes" + "context" + "encoding/json" + "io" + "net" + "net/http" + "strings" + "sync" + "testing" + "time" + + omni "git.yuntuo.hk/k2safe/go-omni" + "git.yuntuo.hk/k2safe/go-omni/auth" + "git.yuntuo.hk/k2safe/go-omni/examples/reference/repository" + "git.yuntuo.hk/k2safe/go-omni/internal/testutil" +) + +// networkHandler runs the same business assertions through a real TCP client. +// The application handler is only called by its configured server transport. +type networkHandler struct { + t *testing.T + client *http.Client + address, scheme, token, alpn string + proto int + requests *int +} + +func (h networkHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + h.t.Helper() + req := r.Clone(r.Context()) + req.RequestURI = "" + req.URL.Scheme, req.URL.Host, req.Host = h.scheme, h.address, h.address + if h.token != "" && req.Header.Get("Authorization") == "" { + req.Header.Set("Authorization", "Bearer "+h.token) + } + response, err := h.client.Do(req) + if err != nil { + h.t.Fatal(err) + } + defer response.Body.Close() + if strings.HasPrefix(req.URL.Path, "/api/v1/") { + *h.requests++ + } + if response.ProtoMajor != h.proto { + h.t.Fatalf("protocol %s, want HTTP/%d", response.Proto, h.proto) + } + if h.scheme == "https" && (response.TLS == nil || len(response.TLS.VerifiedChains) == 0 || response.TLS.NegotiatedProtocol != h.alpn) { + h.t.Fatalf("TLS/ALPN verification failed: %+v", response.TLS) + } + for key, values := range response.Header { + w.Header()[key] = append([]string(nil), values...) + } + w.WriteHeader(response.StatusCode) + if _, err := io.Copy(w, response.Body); err != nil { + h.t.Fatal(err) + } +} + +type lockedLog struct { + sync.Mutex + bytes.Buffer +} + +func (b *lockedLog) Write(p []byte) (int, error) { + b.Lock() + defer b.Unlock() + return b.Buffer.Write(p) +} + +func (b *lockedLog) snapshot() string { + b.Lock() + defer b.Unlock() + return b.Buffer.String() +} + +func TestBusinessConfiguredTransports(t *testing.T) { + for _, test := range []struct { + name, transport string + tls bool + proto int + alpn string + }{ + {"nethttp", omni.TransportNetHTTP, false, 1, ""}, + {"native", omni.TransportNative, false, 1, ""}, + {"nethttpTLS", omni.TransportNetHTTP, true, 2, "h2"}, + {"nativeTLS", omni.TransportNative, true, 1, "http/1.1"}, + {"http2TLS", omni.TransportHTTP2, true, 2, "h2"}, + } { + t.Run(test.name, func(t *testing.T) { + cert, key, tlsClient := testutil.CertificateFiles(t) + t.Setenv("REFERENCE_TEST_TRANSPORT", test.transport) + t.Setenv("REFERENCE_TEST_SHUTDOWN_TIMEOUT", "3s") + t.Setenv("REFERENCE_TEST_TLS_CERT_FILE", "") + t.Setenv("REFERENCE_TEST_TLS_KEY_FILE", "") + if test.tls { + t.Setenv("REFERENCE_TEST_TLS_CERT_FILE", cert) + t.Setenv("REFERENCE_TEST_TLS_KEY_FILE", key) + } + settings, err := omni.LoadConfig("REFERENCE_TEST_") + if err != nil { + t.Fatal(err) + } + var logs lockedLog + logger := testLogger(&logs) + application := newProtectedApp(repository.NewMemoryUsers(), logger, auth.StaticAuthenticator{ + "admin": {UserID: "admin", Roles: []string{"admin"}}, + "user": {UserID: "user", Roles: []string{"user"}}, + }, omni.WithConfig(settings)) + application.OnShutdown(logger.Close) + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- application.ServeContext(ctx, listener) }() + t.Cleanup(func() { + cancel() + select { + case err := <-done: + if err != nil { + t.Error(err) + } + case <-time.After(5 * time.Second): + t.Error("business server did not stop") + listener.Close() + } + }) + client := &http.Client{Timeout: 5 * time.Second, Transport: &http.Transport{TLSClientConfig: tlsClient, ForceAttemptHTTP2: true}} + defer client.CloseIdleConnections() + scheme := "http" + if test.tls { + scheme = "https" + } + requests := 0 + network := networkHandler{t: t, client: client, address: listener.Addr().String(), scheme: scheme, proto: test.proto, alpn: test.alpn, requests: &requests} + request(t, network, "POST", "/api/v1/users", `{"name":"Alice","email":"alice@example.com"}`, 401, "UNAUTHENTICATED") + requestWithAuthorization(t, network, "DELETE", "/api/v1/users/1", "", "Bearer user", 403, "FORBIDDEN") + network.token = "admin" + checkUserCRUD(t, network) + request(t, network, "POST", "/api/v1/users", `{"name":`, 400, "INVALID_BODY") + count := 0 + for _, line := range strings.Split(strings.TrimSpace(logs.snapshot()), "\n") { + var record map[string]any + if err := json.Unmarshal([]byte(line), &record); err != nil { + t.Fatal(err) + } + if record["requestId"] == "reference-test" { + count++ + } + } + if count < requests { + t.Fatalf("missing correlated business logs: %d", count) + } + }) + } +} + +func TestReferenceAddressCompatibility(t *testing.T) { + t.Setenv("GO_OMNI_TRANSPORT", "native") + t.Setenv("GO_OMNI_TLS_CERT_FILE", "") + t.Setenv("GO_OMNI_TLS_KEY_FILE", "") + t.Setenv("GO_OMNI_HTTP_ADDR", "") + t.Setenv("GO_OMNI_ADDR", "") + settings, err := referenceConfig() + if err != nil || settings.Address != "127.0.0.1:8080" || settings.Transport != "native" { + t.Fatalf("default: %+v %v", settings, err) + } + t.Setenv("GO_OMNI_ADDR", "127.0.0.1:8081") + settings, err = referenceConfig() + if err != nil || settings.Address != "127.0.0.1:8081" { + t.Fatalf("legacy: %+v %v", settings, err) + } + t.Setenv("GO_OMNI_HTTP_ADDR", "127.0.0.1:8082") + settings, err = referenceConfig() + if err != nil || settings.Address != "127.0.0.1:8082" { + t.Fatalf("standard: %+v %v", settings, err) + } +} diff --git a/internal/testutil/tls.go b/internal/testutil/tls.go new file mode 100644 index 0000000..3b44647 --- /dev/null +++ b/internal/testutil/tls.go @@ -0,0 +1,61 @@ +// Package testutil contains fixtures shared by transport integration tests. +package testutil + +import ( + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/tls" + "crypto/x509" + "encoding/pem" + "math/big" + "net" + "os" + "path/filepath" + "testing" + "time" +) + +// CertificateFiles creates a temporary localhost certificate and a client +// configuration that verifies it. No insecure TLS bypass is needed in tests. +func CertificateFiles(t testing.TB) (certFile, keyFile string, client *tls.Config) { + t.Helper() + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + t.Fatal(err) + } + serial, err := rand.Int(rand.Reader, new(big.Int).Lsh(big.NewInt(1), 120)) + if err != nil { + t.Fatal(err) + } + certificate := &x509.Certificate{ + SerialNumber: serial, + NotBefore: time.Now().Add(-time.Hour), NotAfter: time.Now().Add(time.Hour), + DNSNames: []string{"localhost"}, IPAddresses: []net.IP{net.ParseIP("127.0.0.1")}, + KeyUsage: x509.KeyUsageDigitalSignature, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, + } + der, err := x509.CreateCertificate(rand.Reader, certificate, certificate, &key.PublicKey, key) + if err != nil { + t.Fatal(err) + } + keyDER, err := x509.MarshalPKCS8PrivateKey(key) + if err != nil { + t.Fatal(err) + } + directory := t.TempDir() + certFile, keyFile = filepath.Join(directory, "server.crt"), filepath.Join(directory, "server.key") + if err := os.WriteFile(certFile, pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}), 0600); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(keyFile, pem.EncodeToMemory(&pem.Block{Type: "PRIVATE KEY", Bytes: keyDER}), 0600); err != nil { + t.Fatal(err) + } + leaf, err := x509.ParseCertificate(der) + if err != nil { + t.Fatal(err) + } + roots := x509.NewCertPool() + roots.AddCert(leaf) + return certFile, keyFile, &tls.Config{RootCAs: roots, MinVersion: tls.VersionTLS12} +} diff --git a/omni.go b/omni.go index 8f50ae1..bfe55f2 100644 --- a/omni.go +++ b/omni.go @@ -25,12 +25,19 @@ type Router = httpx.Router type HTTPRouterGroup = httpx.RouterGroup type APIError = apierrors.APIError type Config = config.Config +type TLSConfig = config.TLSConfig type Option = config.Option type Response = response.Response type LogConfig = logx.Config type Logger = logx.Logger type Metrics = metrics.Metrics +const ( + TransportNetHTTP = config.TransportNetHTTP + TransportNative = config.TransportNative + TransportHTTP2 = config.TransportHTTP2 +) + func New(options ...Option) *App { return app.New(options...) } func NewWithRouter(router Router, options ...Option) *App { @@ -45,6 +52,10 @@ func WithTrustedProxies(proxies ...string) Option { return config.WithTrustedPro func WithConfig(settings Config) Option { return config.WithConfig(settings) } +func WithTransport(name string) Option { return config.WithTransport(name) } + +func WithTLS(certFile, keyFile string) Option { return config.WithTLS(certFile, keyFile) } + func WithLogConfig(settings LogConfig) Option { return config.WithLogConfig(settings) } func WithShutdownTimeout(timeout time.Duration) Option { return config.WithShutdownTimeout(timeout) } diff --git a/scripts/verifyScaffold.py b/scripts/verifyScaffold.py index f87a0b5..e8fbc56 100644 --- a/scripts/verifyScaffold.py +++ b/scripts/verifyScaffold.py @@ -24,12 +24,13 @@ def run(args, cwd, env): return result.stdout -def exerciseServer(binary, directory, env, stopSignal, cleanupMarker=None): +def exerciseServer(binary, directory, env, stopSignal, cleanupMarker=None, transport="nethttp"): with socket.socket() as listener: listener.bind(("127.0.0.1", 0)) address = f"127.0.0.1:{listener.getsockname()[1]}" serverEnv = {key: value for key, value in env.items() if not key.startswith("GO_OMNI_")} serverEnv.update(GO_OMNI_HTTP_ADDR=address, + GO_OMNI_TRANSPORT=transport, GO_OMNI_ENABLE_LOGGER="true", GO_OMNI_SHUTDOWN_TIMEOUT="2s") if cleanupMarker: serverEnv["GO_OMNI_CLEANUP_MARKER"] = str(cleanupMarker) @@ -85,7 +86,7 @@ def exerciseServer(binary, directory, env, stopSignal, cleanupMarker=None): if process.poll() is None: process.kill() process.wait(timeout=5) - print(f"PASS {binary.name}: HTTP checks, {stopSignal.name}, exit 0, listener closed", flush=True) + print(f"PASS {binary.name} ({transport}): HTTP checks, {stopSignal.name}, exit 0, listener closed", flush=True) def main(): @@ -95,6 +96,7 @@ def main(): parser.add_argument("--cli", help="Use a locally built CLI instead of installing") parser.add_argument("--framework-version", help="Explicit version passed to the CLI") parser.add_argument("--framework-dir", help="Local checkout for pre-release verification") + parser.add_argument("--transport", choices=("nethttp", "native"), default="nethttp", help="Plain HTTP transport to exercise (native requires a framework supporting unified transport configuration)") parser.add_argument("--skip-run-check", action="store_true", help="Skip core Run signal checks when testing an older framework") args = parser.parse_args() env = dict(os.environ, GOWORK="off") @@ -128,7 +130,7 @@ def main(): binary = directory / "server" run([args.go, "build", "-o", str(binary), "./cmd/server"], project, env) for stopSignal in (signal.SIGINT, signal.SIGTERM): - exerciseServer(binary, project, env, stopSignal) + exerciseServer(binary, project, env, stopSignal, transport=args.transport) if not args.skip_run_check: # Verify the framework's convenient Run API independently from the @@ -142,7 +144,9 @@ import ( omni "git.yuntuo.hk/k2safe/go-omni" ) func main() { - app := omni.New() + settings, err := omni.LoadConfig("") + if err != nil { log.Fatal(err) } + app := omni.New(omni.WithConfig(settings)) app.GET("/health/live", func(ctx *omni.Context) error { return ctx.OK(map[string]string{"status": "ok"}) }) app.OnShutdown(func() error { return os.WriteFile(os.Getenv("GO_OMNI_CLEANUP_MARKER"), []byte("closed"), 0600) }) if err := app.Run(os.Getenv("GO_OMNI_HTTP_ADDR")); err != nil { log.Fatal(err) } @@ -152,7 +156,7 @@ func main() { run([args.go, "build", "-o", str(runBinary), "./cmd/runCheck"], project, env) for stopSignal in (signal.SIGINT, signal.SIGTERM): marker = project / f"cleanup-{stopSignal.name}" - exerciseServer(runBinary, project, env, stopSignal, marker) + exerciseServer(runBinary, project, env, stopSignal, marker, transport=args.transport) if __name__ == "__main__": diff --git a/transport/native/HTTP2.md b/transport/native/HTTP2.md index 174578f..223fc3f 100644 --- a/transport/native/HTTP2.md +++ b/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。 diff --git a/transport/native/server.go b/transport/native/server.go index f0d0949..bd080bc 100644 --- a/transport/native/server.go +++ b/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) diff --git a/transport/native/server_test.go b/transport/native/server_test.go index b6dbc2a..7900856 100644 --- a/transport/native/server_test.go +++ b/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: