From ee7fc85031a07674571d07c1019acfbbd07440a9 Mon Sep 17 00:00:00 2001 From: youfak Date: Wed, 29 Jul 2026 11:31:12 +0800 Subject: [PATCH] feat: assemble controller HTTP runtime --- docs/api/admin.md | 2 + docs/api/distribution.md | 3 +- docs/design/project-structure.md | 5 +- docs/development/implementation-plan.md | 6 +- docs/requirements/completion-audit.md | 5 +- internal/controller/runtime/runtime.go | 167 ++++++++++++++++ internal/controller/runtime/runtime_test.go | 199 ++++++++++++++++++++ internal/platform/httpserver/server.go | 197 +++++++++++++++++++ internal/platform/httpserver/server_test.go | 158 ++++++++++++++++ progress.md | 10 + 10 files changed, 744 insertions(+), 8 deletions(-) create mode 100644 internal/controller/runtime/runtime.go create mode 100644 internal/controller/runtime/runtime_test.go create mode 100644 internal/platform/httpserver/server.go create mode 100644 internal/platform/httpserver/server_test.go diff --git a/docs/api/admin.md b/docs/api/admin.md index eaa69ef..36040ba 100644 --- a/docs/api/admin.md +++ b/docs/api/admin.md @@ -28,6 +28,8 @@ Admin API 使用独立监听器与权限，契约位于 `api/openapi/admin.yaml` `platform/httpapi` 公用实现提供。Admin Handler 必须注入 `Authorizer`，标准 装配使用 `httpsecurity.Protection`，并在路由匹配前完成保护。网关使用的 `Proxy-Authorization`/407 语义不得复用到 Admin 的 `Authorization`/401 语义。 +`controller/runtime` 将 Admin 与 Distribution 放在不同 `net.Listener`，任一 +监听器异常会触发同组端点的有界优雅停机。 除契约中的 401/403/404/409/422 外，运行时还明确返回： diff --git a/docs/api/distribution.md b/docs/api/distribution.md index 9c3ab0e..10f4d3b 100644 --- a/docs/api/distribution.md +++ b/docs/api/distribution.md @@ -212,4 +212,5 @@ proxyId, clientId, sourceIP, requestId, upstream, extractedAt, expiresAt 请求体解码、Request ID 与 Problem JSON 统一复用 `platform/httpapi`。认证、 可信代理、来源控制、Client ID 和准入限流由必需的 `IdentityResolver` 注入， 标准装配使用 `httpsecurity.Protection`；解析结果至少包含稳定 Client ID 或 -Source IP，且安全检查先于请求体解析。 +Source IP，且安全检查先于请求体解析。`controller/runtime` 将 Distribution +与 Admin 装配到不同 `net.Listener`，健康端点保持公开，提取端点执行独立认证。 diff --git a/docs/design/project-structure.md b/docs/design/project-structure.md index 88e53ad..a2b1275 100644 --- a/docs/design/project-structure.md +++ b/docs/design/project-structure.md @@ -13,9 +13,9 @@ proxy-pool/ │ ├── config/ # 严格配置解析和校验 │ ├── domain/ # 无传输、无存储依赖的领域模型 │ ├── gateway/ # snapshot、dispatch、server、transport -│ ├── controller/ # provider、pool、routing、extraction、health +│ ├── controller/ # provider、pool、routing、extraction、health、runtime │ ├── adapters/ # PostgreSQL、Redis、Provider API、内存适配 -│ └── platform/ # 日志、指标、停机和进程装配 +│ └── platform/ # HTTP、安全、日志、指标、停机和进程装配 ├── api/ # OpenAPI 与 Protobuf 契约 ├── configs/ # 默认配置 ├── examples/ # 可校验配置场景 @@ -93,4 +93,3 @@ platform -------------------------> standard library / observability SDK 执行方式时再抽象 egress adapter。 - 新路由策略：实现同一策略端口，并提供确定性单测和并发不变量测试。 - 新存储：实现已有 repository port，不把驱动类型泄漏到控制器。 - diff --git a/docs/development/implementation-plan.md b/docs/development/implementation-plan.md index 1ecb9f2..0809bfc 100644 --- a/docs/development/implementation-plan.md +++ b/docs/development/implementation-plan.md @@ -168,8 +168,10 @@ extract/live/ready Handler 与 Admin status/enable/disable/switch/reload Handler 定向契约测试已覆盖严格 JSON、Body 上限、Request ID、幂等 Header、DTO 映射、 404/405 及业务错误映射。共享 `platform/httpsecurity` 已补齐 Basic/API Key/ Bearer/CIDR、可信代理、Client ID、本地准入和 API 401/Gateway 407 差异，并作为 -Admin/Distribution 必需依赖。端点正式勾选仍等待独立监听器装配、 -PostgreSQL/Redis Adapter 与 Compose 集成测试。 +Admin/Distribution 必需依赖。共享 `platform/httpserver` 与 +`controller/runtime` 已完成 Distribution/Admin 独立监听器、首错联动关闭和 +有界优雅停机；端点正式勾选仍等待 PostgreSQL/Redis Adapter、命令入口与 +Compose 集成测试。 ## Task 11: Checker and Health Reducer diff --git a/docs/requirements/completion-audit.md b/docs/requirements/completion-audit.md index 5619d05..a8e8a7c 100644 --- a/docs/requirements/completion-audit.md +++ b/docs/requirements/completion-audit.md @@ -38,7 +38,8 @@ - `PROVIDER-*`：Provider HTTP Client、严格响应上限、模板解析安全边界、凭据 引用 Store 与 Reconciler Adapter 已实现。 - `DIST/Admin HTTP`：严格 JSON、Request ID、Problem 响应及 Distribution/Admin - Handler 已实现；共享认证、CIDR、可信代理、Client ID 与本地准入保护链已接入。 + Handler 已实现；共享认证、CIDR、可信代理、Client ID 与本地准入保护链已接入， + Controller Runtime 已将二者装配到独立监听器并支持联动优雅停机。 ## 2. 已执行验证 @@ -68,7 +69,7 @@ CI 已配置 Linux race job。Docker/Kubernetes 仅完成静态验证，没有 5. Redis Leader、速率限制、心跳与可重建协调适配器。 6. Worker ownership drain/ACK/过期回收和网络快照流。 7. Checker 调度、探测器和健康 reducer。 -8. Admin/Distribution 独立监听器装配、细粒度授权、分布式限流和审计查询。 +8. Admin/Distribution 细粒度授权、分布式限流和审计查询。 9. 真实 Compose/Kubernetes 集成、故障演练和代表性集群负载测试。 ## 4. 容量结论 diff --git a/internal/controller/runtime/runtime.go b/internal/controller/runtime/runtime.go new file mode 100644 index 0000000..608a975 --- /dev/null +++ b/internal/controller/runtime/runtime.go @@ -0,0 +1,167 @@ +package runtime + +import ( + "context" + "errors" + "fmt" + "net" + "net/http" + "strings" + + "github.com/proxy-pool/proxy-pool/internal/config" + "github.com/proxy-pool/proxy-pool/internal/controller/admin" + "github.com/proxy-pool/proxy-pool/internal/controller/distribution" + "github.com/proxy-pool/proxy-pool/internal/platform/httpsecurity" + "github.com/proxy-pool/proxy-pool/internal/platform/httpserver" +) + +const ( + defaultDistributionBodyLimitBytes int64 = 64 << 10 + defaultAdminBodyLimitBytes int64 = 8 << 10 +) + +var ErrInvalidRuntime = errors.New("invalid controller HTTP runtime") + +type Dependencies struct { + Extractor distribution.Extractor + Readiness distribution.ReadinessChecker + AdminService admin.Service +} + +type Options struct { + DistributionBodyLimitBytes int64 + AdminBodyLimitBytes int64 + HTTP httpserver.Options +} + +type Listeners struct { + Distribution net.Listener + Admin net.Listener +} + +type Runtime struct { + distributionEnabled bool + distributionAddress string + distributionHandler http.Handler + adminEnabled bool + adminAddress string + adminHandler http.Handler + httpOptions httpserver.Options +} + +func New(cfg *config.Config, dependencies Dependencies, options Options) (*Runtime, error) { + if cfg == nil || (!cfg.Distribution.Enabled && !cfg.Admin.Enabled) { + return nil, ErrInvalidRuntime + } + distributionBodyLimit, adminBodyLimit, err := resolveBodyLimits(options) + if err != nil { + return nil, err + } + result := &Runtime{httpOptions: options.HTTP} + + if cfg.Distribution.Enabled { + if strings.TrimSpace(cfg.Distribution.Listen) == "" || dependencies.Extractor == nil || dependencies.Readiness == nil { + return nil, ErrInvalidRuntime + } + protection, err := httpsecurity.BuildFromListener( + cfg.Distribution.Listener, + cfg.Distribution.ClientIdentification.Mode, + httpsecurity.APIAuthSemantics, + ) + if err != nil { + return nil, fmt.Errorf("%w: build distribution protection: %v", ErrInvalidRuntime, err) + } + handler, err := distribution.NewHandler(distribution.Config{BodyLimitBytes: distributionBodyLimit}, distribution.Dependencies{ + Extractor: dependencies.Extractor, + Identity: protection, + Readiness: dependencies.Readiness, + }) + if err != nil { + return nil, fmt.Errorf("%w: %v", ErrInvalidRuntime, err) + } + result.distributionEnabled = true + result.distributionAddress = cfg.Distribution.Listen + result.distributionHandler = handler + } + + if cfg.Admin.Enabled { + if strings.TrimSpace(cfg.Admin.Listen) == "" || dependencies.AdminService == nil { + return nil, ErrInvalidRuntime + } + protection, err := httpsecurity.BuildFromListener( + cfg.Admin, + httpsecurity.ClientSourceIP, + httpsecurity.APIAuthSemantics, + ) + if err != nil { + return nil, fmt.Errorf("%w: build admin protection: %v", ErrInvalidRuntime, err) + } + handler, err := admin.NewHandler(dependencies.AdminService, protection, admin.Options{MaxBodyBytes: adminBodyLimit}) + if err != nil { + return nil, fmt.Errorf("%w: %v", ErrInvalidRuntime, err) + } + result.adminEnabled = true + result.adminAddress = cfg.Admin.Listen + result.adminHandler = handler + } + return result, nil +} + +func (runtime *Runtime) Run(ctx context.Context) error { + if runtime == nil || ctx == nil { + return ErrInvalidRuntime + } + bindings := make([]httpserver.Binding, 0, 2) + if runtime.distributionEnabled { + bindings = append(bindings, httpserver.Binding{ + Name: "distribution", Address: runtime.distributionAddress, Handler: runtime.distributionHandler, + }) + } + if runtime.adminEnabled { + bindings = append(bindings, httpserver.Binding{ + Name: "admin", Address: runtime.adminAddress, Handler: runtime.adminHandler, + }) + } + if err := httpserver.ListenAndServe(ctx, runtime.httpOptions, bindings...); err != nil { + return fmt.Errorf("run controller HTTP runtime: %w", err) + } + return nil +} + +func (runtime *Runtime) Serve(ctx context.Context, listeners Listeners) error { + if runtime == nil || ctx == nil || + runtime.distributionEnabled != (listeners.Distribution != nil) || + runtime.adminEnabled != (listeners.Admin != nil) { + return ErrInvalidRuntime + } + endpoints := make([]httpserver.Endpoint, 0, 2) + if runtime.distributionEnabled { + endpoints = append(endpoints, httpserver.Endpoint{ + Name: "distribution", Listener: listeners.Distribution, Handler: runtime.distributionHandler, + }) + } + if runtime.adminEnabled { + endpoints = append(endpoints, httpserver.Endpoint{ + Name: "admin", Listener: listeners.Admin, Handler: runtime.adminHandler, + }) + } + if err := httpserver.Serve(ctx, runtime.httpOptions, endpoints...); err != nil { + return fmt.Errorf("serve controller HTTP runtime: %w", err) + } + return nil +} + +func resolveBodyLimits(options Options) (int64, int64, error) { + if options.DistributionBodyLimitBytes < 0 || options.AdminBodyLimitBytes < 0 { + return 0, 0, ErrInvalidRuntime + } + distributionLimit := options.DistributionBodyLimitBytes + if distributionLimit == 0 { + distributionLimit = defaultDistributionBodyLimitBytes + } + adminLimit := options.AdminBodyLimitBytes + if adminLimit == 0 { + adminLimit = defaultAdminBodyLimitBytes + } + return distributionLimit, adminLimit, nil +} diff --git a/internal/controller/runtime/runtime_test.go b/internal/controller/runtime/runtime_test.go new file mode 100644 index 0000000..754a14c --- /dev/null +++ b/internal/controller/runtime/runtime_test.go @@ -0,0 +1,199 @@ +package runtime + +import ( + "context" + "errors" + "io" + "net" + "net/http" + "sync/atomic" + "testing" + "time" + + "github.com/proxy-pool/proxy-pool/internal/config" + "github.com/proxy-pool/proxy-pool/internal/controller/admin" + controllerExtraction "github.com/proxy-pool/proxy-pool/internal/controller/extraction" + "github.com/proxy-pool/proxy-pool/internal/platform/httpserver" +) + +func TestRuntimeServesDistributionAndAdminOnIndependentListeners(t *testing.T) { + t.Parallel() + cfg := runtimeConfig() + adminService := &stubAdminService{status: admin.Status{ConfigVersion: "cfg-7", SnapshotVersion: 11}} + runtime, err := New(cfg, Dependencies{ + Extractor: stubExtractor{}, + Readiness: stubReadiness{}, + AdminService: adminService, + }, Options{HTTP: testHTTPOptions()}) + if err != nil { + t.Fatalf("New() error = %v", err) + } + distributionListener := mustListen(t) + adminListener := mustListen(t) + ctx, cancel := context.WithCancel(context.Background()) + result := make(chan error, 1) + go func() { + result <- runtime.Serve(ctx, Listeners{ + Distribution: distributionListener, + Admin: adminListener, + }) + }() + + distributionURL := "http://" + distributionListener.Addr().String() + adminURL := "http://" + adminListener.Addr().String() + assertStatus(t, http.MethodGet, distributionURL+"/health/live", nil, http.StatusOK) + assertStatus(t, http.MethodGet, distributionURL+"/api/v1/status", nil, http.StatusNotFound) + assertStatus(t, http.MethodGet, adminURL+"/api/v1/status", nil, http.StatusUnauthorized) + adminHeaders := http.Header{"Authorization": []string{"Bearer admin-token"}} + assertStatus(t, http.MethodGet, adminURL+"/api/v1/status", adminHeaders, http.StatusOK) + assertStatus(t, http.MethodGet, adminURL+"/health/live", adminHeaders, http.StatusNotFound) + if adminService.statusCalls.Load() != 1 { + t.Fatalf("admin status calls = %d, want 1", adminService.statusCalls.Load()) + } + + cancel() + select { + case err := <-result: + if err != nil { + t.Fatalf("Serve() error = %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("Serve() did not stop after cancellation") + } +} + +func TestNewRejectsInvalidRuntimeDependencies(t *testing.T) { + t.Parallel() + validDependencies := Dependencies{ + Extractor: stubExtractor{}, Readiness: stubReadiness{}, AdminService: &stubAdminService{}, + } + tests := []struct { + name string + config *config.Config + deps Dependencies + options Options + }{ + {name: "nil config", deps: validDependencies}, + {name: "no HTTP listener", config: &config.Config{}, deps: validDependencies}, + {name: "distribution extractor", config: runtimeConfig(), deps: Dependencies{Readiness: stubReadiness{}, AdminService: &stubAdminService{}}}, + {name: "distribution readiness", config: runtimeConfig(), deps: Dependencies{Extractor: stubExtractor{}, AdminService: &stubAdminService{}}}, + {name: "admin service", config: runtimeConfig(), deps: Dependencies{Extractor: stubExtractor{}, Readiness: stubReadiness{}}}, + {name: "negative distribution body limit", config: runtimeConfig(), deps: validDependencies, options: Options{DistributionBodyLimitBytes: -1}}, + {name: "negative admin body limit", config: runtimeConfig(), deps: validDependencies, options: Options{AdminBodyLimitBytes: -1}}, + } + for _, test := range tests { + test := test + t.Run(test.name, func(t *testing.T) { + if _, err := New(test.config, test.deps, test.options); !errors.Is(err, ErrInvalidRuntime) { + t.Fatalf("New() error = %v, want %v", err, ErrInvalidRuntime) + } + }) + } +} + +func TestServeRequiresExactlyTheEnabledListeners(t *testing.T) { + t.Parallel() + runtime, err := New(runtimeConfig(), Dependencies{ + Extractor: stubExtractor{}, Readiness: stubReadiness{}, AdminService: &stubAdminService{}, + }, Options{}) + if err != nil { + t.Fatalf("New() error = %v", err) + } + listener := mustListen(t) + defer listener.Close() + + if err := runtime.Serve(context.Background(), Listeners{Distribution: listener}); !errors.Is(err, ErrInvalidRuntime) { + t.Fatalf("Serve() error = %v, want %v", err, ErrInvalidRuntime) + } +} + +func runtimeConfig() *config.Config { + return &config.Config{ + Distribution: config.Distribution{ + Listener: config.Listener{ + Enabled: true, + Listen: "127.0.0.1:0", + Auth: config.Auth{ + Mode: "apiKey", Header: "X-API-Key", Token: "distribution-token", + }, + }, + ClientIdentification: config.ClientIdentification{Mode: "authenticatedClient"}, + }, + Admin: config.Listener{ + Enabled: true, + Listen: "127.0.0.1:0", + Auth: config.Auth{Mode: "bearer", Token: "admin-token"}, + }, + } +} + +func testHTTPOptions() httpserver.Options { + return httpserver.Options{ + ReadHeaderTimeout: time.Second, + ReadTimeout: time.Second, + WriteTimeout: time.Second, + IdleTimeout: time.Second, + ShutdownTimeout: time.Second, + MaxHeaderBytes: 16 << 10, + } +} + +func mustListen(t *testing.T) net.Listener { + t.Helper() + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("Listen(): %v", err) + } + return listener +} + +func assertStatus(t *testing.T, method, target string, headers http.Header, want int) { + t.Helper() + request, err := http.NewRequest(method, target, nil) + if err != nil { + t.Fatalf("NewRequest(): %v", err) + } + request.Header = headers.Clone() + client := &http.Client{Timeout: time.Second} + response, err := client.Do(request) + if err != nil { + t.Fatalf("Do(%s): %v", target, err) + } + defer response.Body.Close() + if response.StatusCode != want { + body, _ := io.ReadAll(response.Body) + t.Fatalf("%s status = %d, want %d; body=%s", target, response.StatusCode, want, body) + } +} + +type stubExtractor struct{} + +func (stubExtractor) Extract(context.Context, controllerExtraction.Request) (controllerExtraction.Response, error) { + return controllerExtraction.Response{}, nil +} + +type stubReadiness struct{} + +func (stubReadiness) Ready(context.Context) error { return nil } + +type stubAdminService struct { + status admin.Status + statusCalls atomic.Int64 +} + +func (service *stubAdminService) Status(context.Context) (admin.Status, error) { + service.statusCalls.Add(1) + return service.status, nil +} + +func (*stubAdminService) SetUpstreamEnabled(context.Context, admin.SetUpstreamCommand) (admin.MutationResult, error) { + return admin.MutationResult{}, nil +} + +func (*stubAdminService) SwitchRouting(context.Context, admin.SwitchCommand) (admin.MutationResult, error) { + return admin.MutationResult{}, nil +} + +func (*stubAdminService) ReloadConfiguration(context.Context, admin.ReloadCommand) (admin.MutationResult, error) { + return admin.MutationResult{}, nil +} diff --git a/internal/platform/httpserver/server.go b/internal/platform/httpserver/server.go new file mode 100644 index 0000000..2e4e807 --- /dev/null +++ b/internal/platform/httpserver/server.go @@ -0,0 +1,197 @@ +package httpserver + +import ( + "context" + "errors" + "fmt" + "net" + "net/http" + "sync" + "time" +) + +var ( + ErrInvalidEndpoint = errors.New("invalid HTTP endpoint") + ErrInvalidOptions = errors.New("invalid HTTP server options") +) + +type Options struct { + ReadHeaderTimeout time.Duration + ReadTimeout time.Duration + WriteTimeout time.Duration + IdleTimeout time.Duration + ShutdownTimeout time.Duration + MaxHeaderBytes int +} + +type Endpoint struct { + Name string + Listener net.Listener + Handler http.Handler +} + +type Binding struct { + Name string + Address string + Handler http.Handler +} + +func DefaultOptions() Options { + return Options{ + ReadHeaderTimeout: 5 * time.Second, + ReadTimeout: 15 * time.Second, + WriteTimeout: 30 * time.Second, + IdleTimeout: 60 * time.Second, + ShutdownTimeout: 15 * time.Second, + MaxHeaderBytes: 16 << 10, + } +} + +func Serve(ctx context.Context, options Options, endpoints ...Endpoint) error { + if ctx == nil || len(endpoints) == 0 { + return ErrInvalidEndpoint + } + resolved, err := resolveOptions(options) + if err != nil { + return err + } + seen := make(map[string]struct{}, len(endpoints)) + servers := make([]*http.Server, 0, len(endpoints)) + for _, endpoint := range endpoints { + if endpoint.Name == "" || endpoint.Listener == nil || endpoint.Handler == nil { + return ErrInvalidEndpoint + } + if _, exists := seen[endpoint.Name]; exists { + return fmt.Errorf("%w: duplicate name %q", ErrInvalidEndpoint, endpoint.Name) + } + seen[endpoint.Name] = struct{}{} + servers = append(servers, &http.Server{ + Handler: endpoint.Handler, + ReadHeaderTimeout: resolved.ReadHeaderTimeout, + ReadTimeout: resolved.ReadTimeout, + WriteTimeout: resolved.WriteTimeout, + IdleTimeout: resolved.IdleTimeout, + MaxHeaderBytes: resolved.MaxHeaderBytes, + }) + } + + type serveResult struct { + name string + err error + } + results := make(chan serveResult, len(endpoints)) + for index, endpoint := range endpoints { + server := servers[index] + go func() { + results <- serveResult{name: endpoint.Name, err: server.Serve(endpoint.Listener)} + }() + } + + var firstErr error + received := 0 + select { + case <-ctx.Done(): + case result := <-results: + received++ + if !errors.Is(result.err, http.ErrServerClosed) { + firstErr = fmt.Errorf("serve HTTP endpoint %q: %w", result.name, result.err) + } + } + + shutdownCtx, cancel := context.WithTimeout(context.Background(), resolved.ShutdownTimeout) + defer cancel() + shutdownErrors := make(chan error, len(servers)) + var wait sync.WaitGroup + for _, server := range servers { + wait.Add(1) + go func() { + defer wait.Done() + if shutdownErr := server.Shutdown(shutdownCtx); shutdownErr != nil { + _ = server.Close() + shutdownErrors <- shutdownErr + } + }() + } + wait.Wait() + close(shutdownErrors) + if firstErr == nil { + for shutdownErr := range shutdownErrors { + if firstErr == nil { + firstErr = fmt.Errorf("shutdown HTTP endpoints: %w", shutdownErr) + } + } + } + + for received < len(endpoints) { + result := <-results + received++ + if firstErr == nil && !errors.Is(result.err, http.ErrServerClosed) { + firstErr = fmt.Errorf("serve HTTP endpoint %q: %w", result.name, result.err) + } + } + return firstErr +} + +func ListenAndServe(ctx context.Context, options Options, bindings ...Binding) error { + if ctx == nil || len(bindings) == 0 { + return ErrInvalidEndpoint + } + if _, err := resolveOptions(options); err != nil { + return err + } + seen := make(map[string]struct{}, len(bindings)) + for _, binding := range bindings { + if binding.Name == "" || binding.Address == "" || binding.Handler == nil { + return ErrInvalidEndpoint + } + if _, exists := seen[binding.Name]; exists { + return fmt.Errorf("%w: duplicate name %q", ErrInvalidEndpoint, binding.Name) + } + seen[binding.Name] = struct{}{} + } + endpoints := make([]Endpoint, 0, len(bindings)) + defer func() { closeEndpoints(endpoints) }() + for _, binding := range bindings { + listener, err := (&net.ListenConfig{}).Listen(ctx, "tcp", binding.Address) + if err != nil { + return fmt.Errorf("listen HTTP endpoint %q: %w", binding.Name, err) + } + endpoints = append(endpoints, Endpoint{ + Name: binding.Name, Listener: listener, Handler: binding.Handler, + }) + } + return Serve(ctx, options, endpoints...) +} + +func resolveOptions(options Options) (Options, error) { + defaults := DefaultOptions() + if options.ReadHeaderTimeout == 0 { + options.ReadHeaderTimeout = defaults.ReadHeaderTimeout + } + if options.ReadTimeout == 0 { + options.ReadTimeout = defaults.ReadTimeout + } + if options.WriteTimeout == 0 { + options.WriteTimeout = defaults.WriteTimeout + } + if options.IdleTimeout == 0 { + options.IdleTimeout = defaults.IdleTimeout + } + if options.ShutdownTimeout == 0 { + options.ShutdownTimeout = defaults.ShutdownTimeout + } + if options.MaxHeaderBytes == 0 { + options.MaxHeaderBytes = defaults.MaxHeaderBytes + } + if options.ReadHeaderTimeout < 0 || options.ReadTimeout < 0 || options.WriteTimeout < 0 || + options.IdleTimeout < 0 || options.ShutdownTimeout <= 0 || options.MaxHeaderBytes < 0 { + return Options{}, ErrInvalidOptions + } + return options, nil +} + +func closeEndpoints(endpoints []Endpoint) { + for _, endpoint := range endpoints { + _ = endpoint.Listener.Close() + } +} diff --git a/internal/platform/httpserver/server_test.go b/internal/platform/httpserver/server_test.go new file mode 100644 index 0000000..be17242 --- /dev/null +++ b/internal/platform/httpserver/server_test.go @@ -0,0 +1,158 @@ +package httpserver + +import ( + "context" + "errors" + "io" + "net" + "net/http" + "testing" + "time" +) + +func TestServeRunsIndependentEndpointsAndShutsDownTogether(t *testing.T) { + t.Parallel() + first := mustListen(t) + second := mustListen(t) + ctx, cancel := context.WithCancel(context.Background()) + result := make(chan error, 1) + + go func() { + result <- Serve(ctx, testOptions(), + Endpoint{Name: "distribution", Listener: first, Handler: textHandler("distribution")}, + Endpoint{Name: "admin", Listener: second, Handler: textHandler("admin")}, + ) + }() + + assertBody(t, first.Addr().String(), "distribution") + assertBody(t, second.Addr().String(), "admin") + cancel() + + select { + case err := <-result: + if err != nil { + t.Fatalf("Serve() error = %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("Serve() did not stop after cancellation") + } +} + +func TestServeRejectsInvalidEndpointDefinitions(t *testing.T) { + t.Parallel() + listener := mustListen(t) + defer listener.Close() + + tests := []struct { + name string + endpoints []Endpoint + }{ + {name: "empty"}, + {name: "missing name", endpoints: []Endpoint{{Listener: listener, Handler: textHandler("ok")}}}, + {name: "missing listener", endpoints: []Endpoint{{Name: "api", Handler: textHandler("ok")}}}, + {name: "missing handler", endpoints: []Endpoint{{Name: "api", Listener: listener}}}, + {name: "duplicate name", endpoints: []Endpoint{ + {Name: "api", Listener: listener, Handler: textHandler("one")}, + {Name: "api", Listener: listener, Handler: textHandler("two")}, + }}, + } + for _, test := range tests { + test := test + t.Run(test.name, func(t *testing.T) { + err := Serve(context.Background(), testOptions(), test.endpoints...) + if !errors.Is(err, ErrInvalidEndpoint) { + t.Fatalf("Serve() error = %v, want %v", err, ErrInvalidEndpoint) + } + }) + } +} + +func TestServeStopsSiblingWhenEndpointFails(t *testing.T) { + t.Parallel() + failed := mustListen(t) + sibling := mustListen(t) + failedAddress := failed.Addr().String() + if err := failed.Close(); err != nil { + t.Fatalf("Close(%s): %v", failedAddress, err) + } + + err := Serve(context.Background(), testOptions(), + Endpoint{Name: "failed", Listener: failed, Handler: textHandler("failed")}, + Endpoint{Name: "sibling", Listener: sibling, Handler: textHandler("sibling")}, + ) + if err == nil { + t.Fatal("Serve() error = nil, want endpoint failure") + } + connection, dialErr := net.DialTimeout("tcp", sibling.Addr().String(), 100*time.Millisecond) + if dialErr == nil { + connection.Close() + t.Fatal("sibling listener remained open after endpoint failure") + } +} + +func TestListenAndServeValidatesAllBindingsBeforeOpeningSockets(t *testing.T) { + t.Parallel() + err := ListenAndServe(context.Background(), testOptions(), + Binding{Name: "api", Address: "\x00", Handler: textHandler("one")}, + Binding{Name: "api", Address: "127.0.0.1:0", Handler: textHandler("two")}, + ) + if !errors.Is(err, ErrInvalidEndpoint) { + t.Fatalf("ListenAndServe() error = %v, want %v", err, ErrInvalidEndpoint) + } +} + +func TestListenAndServeTreatsCancellationBeforeBindingAsGracefulStop(t *testing.T) { + t.Parallel() + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + err := ListenAndServe(ctx, testOptions(), Binding{ + Name: "api", Address: "127.0.0.1:0", Handler: textHandler("ok"), + }) + if err != nil { + t.Fatalf("ListenAndServe() error = %v, want nil", err) + } +} + +func testOptions() Options { + return Options{ + ReadHeaderTimeout: time.Second, + ReadTimeout: time.Second, + WriteTimeout: time.Second, + IdleTimeout: time.Second, + ShutdownTimeout: time.Second, + MaxHeaderBytes: 16 << 10, + } +} + +func mustListen(t *testing.T) net.Listener { + t.Helper() + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("Listen(): %v", err) + } + return listener +} + +func textHandler(body string) http.Handler { + return http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) { + _, _ = io.WriteString(writer, body) + }) +} + +func assertBody(t *testing.T, address, want string) { + t.Helper() + client := &http.Client{Timeout: time.Second} + response, err := client.Get("http://" + address) + if err != nil { + t.Fatalf("GET %s: %v", address, err) + } + defer response.Body.Close() + body, err := io.ReadAll(response.Body) + if err != nil { + t.Fatalf("ReadAll(): %v", err) + } + if string(body) != want { + t.Fatalf("body = %q, want %q", body, want) + } +} diff --git a/progress.md b/progress.md index 61ec5c9..91c6ec0 100644 --- a/progress.md +++ b/progress.md @@ -1,5 +1,15 @@ # 项目进度 +## 2026-07-29 + +- 已实现共享 `httpapi`、`httpsecurity` 与 `httpserver`，统一严格 JSON、Problem、 + Request ID、认证、可信代理、Client ID、准入、多监听器生命周期和优雅停机。 +- Distribution/Admin Handler 已装配到独立监听器；Gateway 配置认证统一复用 + `httpsecurity`，API 401 与代理 407 语义保持分离。 +- 部署配置已使用 `${VAR}` 真实环境变量解析，ConfigMap 与本地配置通过 + `LoadResolved` 回归测试。 +- PostgreSQL/Redis Adapter、生产命令入口与代表性 100,000 QPS 集群压测仍待实现。 + ## 2026-07-28 - 用户删除了先前基于不完整网页内容生成的设计文件。