feat: add bounded native http server

CHANGELOG.md

@@ -1,5 +1,12 @@
 # Changelog
 
+## v0.5.0
+
+- 新增 `transport/native` 第一版自研 HTTP/1.1 服务端，接管 TCP 监听、连接、Keep-Alive、请求解析、响应写回和优雅关闭。
+- 新增 `App.RunWith` / `omni.Runner` 启动边界；native 通过标准 `http.Handler` 接入，现有 Handler、Service 和 Repository 无需修改。
+- native 支持 `Content-Length` 请求体、请求头 / 请求体限制、读写超时和服务关闭时的请求取消；通过对象池和 `MaxConnections` / `MaxConcurrentRequests` 控制高并发内存；chunked 请求体暂返回 501。
+- `omni.New()` 默认仍使用只依赖标准库的 `transport/nethttp` 路由，native 作为显式启用的网络层预览实现。
+
 ## v0.4.0
 
 - **破坏性变更**：移除 Gin 适配器及其全部依赖，`omni.New()` 默认使用只依赖标准库的 `transport/nethttp`。

README.md

@@ -5,7 +5,7 @@ GoOmni 是一套面向业务复用的 Go 后端基础框架。当前默认使用
 远程 module 地址是 `git.yuntuo.hk/k2safe/go-omni`。业务项目可以直接引用指定版本：
 
 ```sh
-go get git.yuntuo.hk/k2safe/go-omni@v0.4.0
+go get git.yuntuo.hk/k2safe/go-omni@v0.5.0
 ```
 
 `v0.4.0` 移除了 Gin 和 Gin 专属的 `WithMode` 配置；`omni.New()` 直接使用标准库路由，已有业务 Handler / Service 无需修改。
@@ -13,7 +13,7 @@ go get git.yuntuo.hk/k2safe/go-omni@v0.4.0
 创建新业务项目：
 
 ```sh
-go install git.yuntuo.hk/k2safe/go-omni/cmd/go-omni@v0.4.0
+go install git.yuntuo.hk/k2safe/go-omni/cmd/go-omni@v0.5.0
 go-omni new --module example.com/billing ./billing
 cd billing
 go mod tidy
@@ -43,6 +43,7 @@ go-omni/
 ├── auth/         # 认证和权限接口
 ├── database/     # database/sql 生命周期和事务
 ├── transport/nethttp/ # 标准库网络适配器
+├── transport/native/  # 自研 HTTP/1.1 服务端预览
 ├── examples/     # 示例项目
 └── cmd/          # go-omni 项目脚手架
 ```
@@ -94,6 +95,34 @@ application := omni.NewWithRouter(myRouter)
 
 `app.Engine()` 返回标准 `http.Handler`，`ctx.Raw()` 返回适配器提供的 `http.ResponseWriter`，业务 API 不锁定具体网络实现。
 
+需要试用第一版自研网络层时，业务注册代码无需改变，只替换启动器：
+
+```go
+import (
+	"context"
+	"log"
+	"os"
+	"os/signal"
+	"syscall"
+
+	omni "git.yuntuo.hk/k2safe/go-omni"
+	"git.yuntuo.hk/k2safe/go-omni/transport/native"
+)
+
+ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+defer cancel()
+
+app := omni.New()
+app.GET("/health", func(ctx *omni.Context) error {
+	return ctx.OK(map[string]string{"status": "ok"})
+})
+if err := app.RunWith(ctx, native.New(nil), ":8080"); err != nil {
+	log.Fatal(err)
+}
+```
+
+`transport/native` 当前支持 HTTP/1.1、Keep-Alive、`Content-Length` 请求体、读写超时和优雅关闭；chunked 请求体暂返回 501。它会复用连接状态和响应对象，并可通过 `MaxConnections`、`MaxConcurrentRequests` 限制内存峰值。它仍通过标准 HTTP 请求/响应接口连接业务层，适合在现有 Handler / Service 上做网络层验证。
+
 业务代码可以直接使用 `git.yuntuo.hk/k2safe/go-omni/logx` 打结构化日志，不需要依赖具体网络库：
 
 ```go
@@ -171,10 +200,10 @@ userService := service.NewUsers(services)
 从远程版本生成并升级一个独立业务模块，可运行：
 
 ```sh
-python3 scripts/verifyBusiness.py --version v0.4.0
+python3 scripts/verifyBusiness.py --version v0.5.0
 ```
 
-该验收会在临时目录中先固定 `v0.2.0`，再升级到 `v0.4.0`，检查无本地 replace、配置加载、结构化日志、requestId、CRUD、校验错误、业务错误和优雅退出。
+该验收会在临时目录中先固定 `v0.2.0`，再升级到 `v0.5.0`，检查无本地 replace、配置加载、结构化日志、requestId、CRUD、校验错误、业务错误和优雅退出。
 
 标准库路由和完整 GoOmni 请求链的可重复性能基线见 [PERFORMANCE.md](PERFORMANCE.md)，运行 `python3 scripts/verifyPerformance.py` 可重新采样。
 

ROADMAP.md

@@ -35,10 +35,12 @@
 - 业务试用没有要求修改 Handler / Service；现有公共 API 足以支撑一套分层业务模块。
 - 自动验收入口：`python3 scripts/verifyBusiness.py --version v0.4.0`。
 
-## 5. 稳定性与性能基线（进行中）
+## 5. 自研网络层、稳定性与性能基线（进行中，v0.5.0 预览）
 
 - 已加入标准库路由和完整 GoOmni 请求链 benchmark，输出分配次数和耗时，作为未来自研网络层的对照基线。
-- 已具备全量 race、vet、build 验收；生产压测、长连接和真实部署规模验证仍待完成。
+- 新增 `transport/native` 第一版自研 HTTP/1.1 服务端：自行管理 TCP listener、Keep-Alive、Content-Length 请求解析、响应写回、读写超时和优雅关闭；通过 `App.RunWith` 接入，业务 Handler / Service 保持不变。
+- native 当前拒绝 chunked 请求体，仍使用标准 `http.Request` / `http.ResponseWriter` 作为桥接；解析器、连接抽象和生产压测将在后续阶段继续下沉。
+- 已具备全量 race、vet、build 验收；生产压测、长连接规模和真实部署验证仍待完成。
 - 自动验收入口：`python3 scripts/verifyPerformance.py`。
 
-以上阶段以验收结果判断完成。当前仍是基础框架试用阶段；MySQL、适配层契约和一次业务模块试用已完成，稳定性与性能验证、自研网络内核尚未完成。
+以上阶段以验收结果判断完成。当前仍是基础框架试用阶段；MySQL、适配层契约和一次业务模块试用已完成，自研网络层已有可运行预览，稳定性与生产规模验证尚未完成。

app/app.go

@@ -29,6 +29,13 @@ type App struct {
 	closeErr    error
 }
 
+// Runner is the small lifecycle boundary used by an HTTP server
+// implementation. The default App.RunContext uses net/http; a custom runner
+// can own the listener, protocol parser and connection lifecycle.
+type Runner interface {
+	Run(context.Context, http.Handler, string) error
+}
+
 // New creates an application using GoOmni's standard-library router.
 func New(options ...config.Option) *App {
 	settings := configure(options...)
@@ -150,6 +157,25 @@ func (a *App) RunContext(ctx context.Context, address ...string) error {
 	return a.serveContext(ctx, server, listener)
 }
 
+// RunWith starts a custom server runner while keeping App-owned cleanup in the
+// same lifecycle. The runner receives the transport-neutral App handler.
+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())
+	}
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	listenOn := a.config.Address
+	if listenOn == "" {
+		listenOn = ":8080"
+	}
+	if len(address) > 0 && address[0] != "" {
+		listenOn = address[0]
+	}
+	return errors.Join(runner.Run(ctx, a, listenOn), a.Close())
+}
+
 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) }()

cmd/go-omni/README.md

@@ -3,7 +3,7 @@
 从远程安装，生成一个带 Handler / Service / Repository / Model 分层的服务骨架：
 
 ```sh
-go install git.yuntuo.hk/k2safe/go-omni/cmd/go-omni@v0.4.0
+go install git.yuntuo.hk/k2safe/go-omni/cmd/go-omni@v0.5.0
 go-omni new --module example.com/billing ./billing
 cd billing
 go mod tidy
@@ -16,7 +16,7 @@ go run ./cmd/server
 需要选择其他已发布版本时，使用完整版本号：
 
 ```sh
-go-omni new --framework-version v0.4.0 --module example.com/billing ./billing
+go-omni new --framework-version v0.5.0 --module example.com/billing ./billing
 ```
 
 `--framework-version` 禁用自动本地替换；`main`、`latest` 等浮动引用不接受为固定版本。
@@ -48,8 +48,8 @@ MySQL 项目额外包含 `schema.sql`、MySQL Repository 和可选的数据库
 发布验收使用 Python 3 标准库脚本，不需要额外 Python 依赖：
 
 ```sh
-python3 scripts/verifyScaffold.py --version v0.4.0
-python3 scripts/verifyDatabase.py --version v0.4.0
+python3 scripts/verifyScaffold.py --version v0.5.0
+python3 scripts/verifyDatabase.py --version v0.5.0
 ```
 
 脚本在源码目录外安装远程 CLI、生成项目，确认固定版本且没有 `replace`，再下载依赖、

cmd/go-omni/scaffold/version.go

@@ -10,7 +10,7 @@ const frameworkModule = "git.yuntuo.hk/k2safe/go-omni"
 // Source builds can carry an unpublished VCS version (including +dirty).
 // Keep this fallback in sync with the next release; installed CLIs use their
 // downloaded module version, which also has a module checksum.
-const defaultFrameworkVersion = "v0.4.0"
+const defaultFrameworkVersion = "v0.5.0"
 
 var frameworkVersionPattern = regexp.MustCompile(`^v[01]\.(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)(-((0|[1-9][0-9]*|[0-9]*[A-Za-z-][0-9A-Za-z-]*)(\.(0|[1-9][0-9]*|[0-9]*[A-Za-z-][0-9A-Za-z-]*))*))?(\+[0-9A-Za-z-]+(\.[0-9A-Za-z-]+)*)?$`)
 

omni.go

@@ -13,6 +13,7 @@ import (
 )
 
 type App = app.App
+type Runner = app.Runner
 type RouterGroup = app.RouterGroup
 type Principal = auth.Principal
 type Authenticator = auth.Authenticator

scripts/verifyBusiness.py

@@ -25,7 +25,7 @@ def free_address():
 def main():
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument("--go", default="go")
-    parser.add_argument("--version", default="v0.4.0", help="Target framework release")
+    parser.add_argument("--version", default="v0.5.0", help="Target framework release")
     parser.add_argument("--from-version", default="v0.2.0", help="Version used before upgrade")
     args = parser.parse_args()
 

scripts/verifyDatabase.py

@@ -145,7 +145,7 @@ class Server:
 def main():
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument("--go", default="go")
-    parser.add_argument("--version", default="v0.4.0")
+    parser.add_argument("--version", default="v0.5.0")
     parser.add_argument("--cli", help="Use a locally built CLI")
     parser.add_argument("--framework-dir", help="Local framework source for pre-release checks")
     parser.add_argument("--image", default="mysql:8.4")

scripts/verifyScaffold.py

@@ -91,7 +91,7 @@ def exerciseServer(binary, directory, env, stopSignal, cleanupMarker=None):
 def main():
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument("--go", default="go", help="Go executable")
-    parser.add_argument("--version", default="v0.4.0", help="Released CLI version to install")
+    parser.add_argument("--version", default="v0.5.0", help="Released CLI version to install")
     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")

scripts/verifyTransports.py

@@ -13,7 +13,7 @@ from verifyScaffold import MODULE, run
 def main():
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument("--go", default="go")
-    parser.add_argument("--version", default="v0.4.0")
+    parser.add_argument("--version", default="v0.5.0")
     parser.add_argument("--framework-dir", help="Use a local checkout before release")
     args = parser.parse_args()
     env = dict(os.environ, GOWORK="off")
@@ -55,7 +55,7 @@ func TestContract(t *testing.T) {{
             count = sum(event.get("Action") == "pass" and event.get("Test", "").startswith(name + "/")
                         for event in events)
             print(f"PASS {adapter}: {count} contract checks, including business handlers, cancellation and concurrent requests", flush=True)
-        print("PASS independent module: both adapters accepted by the same public contract suite", flush=True)
+        print("PASS independent module: nethttp accepted by the public contract suite", flush=True)
 
 
 if __name__ == "__main__":

transport/README.md

@@ -14,6 +14,24 @@ registerRoutes(application) // 与 omni.New() 复用同一份业务注册代码
 
 `transport/nethttp` 是当前内置的标准库路由实现。它提供一套可移植的路径、方法和错误回退规则；自研网络层可以直接实现同一个 `httpx.Router` 契约。
 
+`transport/native` 是第一版自研 HTTP/1.1 服务端：它自己管理 TCP listener、连接、Keep-Alive、请求解析、响应写回和优雅关闭，再把请求交给现有 `http.Handler`。当前只支持 `Content-Length` 请求体，chunked 请求会返回 501；它仍使用标准 `http.Request` / `http.ResponseWriter` 作为业务桥接，因此不会影响现有 Handler / Service。
+
+服务端不是把连接交给一个全局串行 worker，而是每条连接独立处理请求；Keep-Alive 负责复用客户端连接，`Server` 内部的 `sync.Pool` 负责复用小型 Reader、响应对象和连接状态。Reader 默认只有 4 KiB，需要处理大请求头时按需增长。需要保护内存上限时，可以设置 `MaxConnections` 和 `MaxConcurrentRequests`；超出连接上限的连接会被关闭，超出请求并发上限的请求返回 503，不会无限排队。
+
+这个设计借鉴了 Nginx 的资源边界，但没有照搬它的进程模型：Nginx 用少量 worker 进程和事件机制管理大量连接，并用 `worker_connections` 限制每个 worker 的同时连接数；Go runtime 自己已经用 epoll / kqueue 等网络轮询器唤醒阻塞的 goroutine，因此 native 采用“accept loop + 每连接轻量 goroutine + 请求并发闸门”的等价实现。这样业务 Handler 可以继续执行同步代码，慢请求不会阻塞其他连接；只有在真实压测证明 goroutine 调度或协议状态机成为瓶颈时，才需要继续下沉为手写事件循环。
+
+使用自研服务端：
+
+```go
+application := omni.New()
+server := native.New(nil)
+if err := application.RunWith(ctx, server, ":8080"); err != nil {
+    log.Fatal(err)
+}
+```
+
+`omni.New()` 默认仍使用 `transport/nethttp` 路由；native 服务端是显式启用的预览实现，等协议和连接边界稳定后再考虑切换默认启动器。
+
 ## 可移植路由规则（v0.3.0）
 
 | 项目 | 约定 |
@@ -67,9 +85,9 @@ func TestContract(t *testing.T) {
 
 ```sh
 go test -race ./transport/... ./examples/reference
-python3 scripts/verifyTransports.py --version v0.4.0
+python3 scripts/verifyTransports.py --version v0.5.0
 ```
 
 第一条验证标准库适配器复用完整 Handler / Service / Repository。第二条在源码目录外创建无 replace 的消费模块，从远程版本导入同一套验收包；自研适配器也应使用同一个公共测试包。
 
-当前替换边界是路由与 HTTP 适配层，`App.Run` 仍使用标准库 `net/http.Server`。未来自研监听、连接管理或协议解析时，可让自研服务调用 `App.ServeHTTP`（提供标准 HTTP 请求/响应桥接），由自研启动层管理监听与关闭，并在退出时调用 `App.Close`；本阶段没有实现新的网络内核。
+当前替换边界分成两层：`transport/nethttp` 提供标准库路由，`transport/native` 进一步接管 TCP 监听、连接管理和 HTTP/1.1 协议 I/O。`App.Run` / `App.RunContext` 仍使用标准库 `net/http.Server`，需要显式调用 `App.RunWith(ctx, native.New(nil), address)` 才会启用 native 预览实现。native 当前通过标准 `http.Request` / `http.ResponseWriter` 桥接到业务层，后续可在不改 Handler / Service 的前提下逐步替换解析器和响应写回。

transport/native/server.go

@@ -0,0 +1,655 @@
+// Package native provides GoOmni's first project-owned HTTP/1.1 server.
+//
+// It deliberately keeps the public handler boundary as net/http so existing
+// GoOmni business code remains unchanged while connection management and
+// protocol I/O move out of net/http.Server. The parser currently supports
+// Content-Length request bodies and rejects chunked requests.
+package native
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"errors"
+	"io"
+	"net"
+	"net/http"
+	"net/url"
+	"sort"
+	"strconv"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+)
+
+const (
+	defaultReadHeaderTimeout = 5 * time.Second
+	defaultReadTimeout       = 30 * time.Second
+	defaultWriteTimeout      = 30 * time.Second
+	defaultIdleTimeout       = 60 * time.Second
+	defaultMaxHeaderBytes    = 1 << 20
+	defaultMaxBodyBytes      = 8 << 20
+	defaultReadBufferSize    = 4 << 10
+	maxPooledResponseBytes   = 64 << 10
+)
+
+// Server is a small HTTP/1.1 server with project-owned connection lifecycle.
+// It supports keep-alive, graceful shutdown, bounded concurrency and standard
+// http.Handler values. Zero MaxConnections and MaxConcurrentRequests mean
+// unlimited; pooled protocol objects are reused automatically.
+type Server struct {
+	ReadHeaderTimeout     time.Duration
+	ReadTimeout           time.Duration
+	WriteTimeout          time.Duration
+	IdleTimeout           time.Duration
+	ReadBufferSize        int
+	MaxHeaderBytes        int
+	MaxBodyBytes          int64
+	MaxConnections        int
+	MaxConcurrentRequests int
+
+	mu             sync.Mutex
+	handler        http.Handler
+	listener       net.Listener
+	conns          map[*connection]struct{}
+	closing        bool
+	cancel         context.CancelFunc
+	wg             sync.WaitGroup
+	connSlots      chan struct{}
+	requestSlots   chan struct{}
+	readerPool     sync.Pool
+	writerPool     sync.Pool
+	connectionPool sync.Pool
+}
+
+type connection struct {
+	server        *Server
+	conn          net.Conn
+	active        atomic.Bool
+	closeWhenIdle atomic.Bool
+	closed        atomic.Bool
+	slotHeld      bool
+	closeDone     chan struct{}
+}
+
+// New creates a native server. The handler may also be supplied later through
+// App.RunWith, which is useful when the App owns the lifecycle.
+func New(handler http.Handler) *Server {
+	server := &Server{
+		handler:           handler,
+		ReadHeaderTimeout: defaultReadHeaderTimeout,
+		ReadTimeout:       defaultReadTimeout,
+		WriteTimeout:      defaultWriteTimeout,
+		IdleTimeout:       defaultIdleTimeout,
+		ReadBufferSize:    defaultReadBufferSize,
+		MaxHeaderBytes:    defaultMaxHeaderBytes,
+		MaxBodyBytes:      defaultMaxBodyBytes,
+		conns:             make(map[*connection]struct{}),
+	}
+	server.readerPool.New = func() any { return bufio.NewReaderSize(nil, server.readBufferSize()) }
+	server.writerPool.New = func() any { return newResponseWriter("") }
+	server.connectionPool.New = func() any { return &connection{server: server} }
+	return server
+}
+
+func (s *Server) readBufferSize() int {
+	if s.ReadBufferSize > 0 {
+		return s.ReadBufferSize
+	}
+	return defaultReadBufferSize
+}
+
+// Run implements app.Runner without importing the app package.
+func (s *Server) Run(ctx context.Context, handler http.Handler, address string) error {
+	if handler != nil {
+		s.handler = handler
+	}
+	if s.handler == nil {
+		return errors.New("go-omni/native: nil handler")
+	}
+	if address == "" {
+		address = ":8080"
+	}
+	listener, err := net.Listen("tcp", address)
+	if err != nil {
+		return err
+	}
+	return s.Serve(ctx, listener)
+}
+
+// Serve accepts connections from listener until the context is canceled or an
+// unrecoverable accept error occurs. The listener is owned by the Server.
+func (s *Server) Serve(ctx context.Context, listener net.Listener) error {
+	if listener == nil {
+		return errors.New("go-omni/native: nil listener")
+	}
+	if s.handler == nil {
+		return errors.New("go-omni/native: nil handler")
+	}
+	if ctx == nil {
+		ctx = context.Background()
+	}
+
+	s.mu.Lock()
+	if s.listener != nil {
+		s.mu.Unlock()
+		return errors.New("go-omni/native: server is already running")
+	}
+	s.listener = listener
+	s.closing = false
+	if s.MaxConnections > 0 {
+		s.connSlots = make(chan struct{}, s.MaxConnections)
+	} else {
+		s.connSlots = nil
+	}
+	if s.MaxConcurrentRequests > 0 {
+		s.requestSlots = make(chan struct{}, s.MaxConcurrentRequests)
+	} else {
+		s.requestSlots = nil
+	}
+	baseCtx, cancel := context.WithCancel(context.Background())
+	s.cancel = cancel
+	s.mu.Unlock()
+	defer cancel()
+
+	shutdownDone := make(chan struct{})
+	go func() {
+		select {
+		case <-ctx.Done():
+			timeout := s.shutdownTimeout()
+			shutdownCtx, cancel := context.WithTimeout(context.Background(), timeout)
+			_ = s.Shutdown(shutdownCtx)
+			cancel()
+		case <-shutdownDone:
+		}
+	}()
+
+	var serveErr error
+	for {
+		conn, err := listener.Accept()
+		if err != nil {
+			if ctx.Err() != nil || s.isClosing() {
+				break
+			}
+			serveErr = err
+			break
+		}
+		if slots := s.connSlots; slots != nil {
+			select {
+			case slots <- struct{}{}:
+			default:
+				_ = conn.Close()
+				continue
+			}
+		}
+		state := s.connectionPool.Get().(*connection)
+		state.server = s
+		state.conn = conn
+		state.active.Store(false)
+		state.closeWhenIdle.Store(false)
+		state.closed.Store(false)
+		state.slotHeld = s.connSlots != nil
+		state.closeDone = make(chan struct{})
+		s.mu.Lock()
+		if s.closing {
+			s.mu.Unlock()
+			state.close()
+			s.recycleConnection(state)
+			continue
+		}
+		s.conns[state] = struct{}{}
+		s.wg.Add(1)
+		s.mu.Unlock()
+		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()
+	timeout := s.shutdownTimeout()
+	shutdownCtx, cancel := context.WithTimeout(context.Background(), timeout)
+	shutdownErr := s.Shutdown(shutdownCtx)
+	cancel()
+	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.
+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()
+	}
+	listener := s.listener
+	states := make([]*connection, 0, len(s.conns))
+	for state := range s.conns {
+		state.closeWhenIdle.Store(true)
+		states = append(states, state)
+	}
+	s.listener = nil
+	s.mu.Unlock()
+	if listener != nil {
+		_ = listener.Close()
+	}
+	for _, state := range states {
+		if !state.active.Load() {
+			state.close()
+		}
+	}
+
+	done := make(chan struct{})
+	go func() {
+		s.wg.Wait()
+		close(done)
+	}()
+	select {
+	case <-done:
+		return nil
+	case <-ctx.Done():
+		for _, state := range states {
+			state.close()
+		}
+		<-done
+		return ctx.Err()
+	}
+}
+
+func (s *Server) shutdownTimeout() time.Duration {
+	if value := s.WriteTimeout; value > 0 {
+		return value
+	}
+	return defaultWriteTimeout
+}
+
+func (s *Server) isClosing() bool {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.closing
+}
+
+func (s *Server) serveConnection(baseCtx context.Context, state *connection) {
+	defer s.wg.Done()
+	defer func() {
+		state.close()
+		s.recycleConnection(state)
+	}()
+	reader := s.readerPool.Get().(*bufio.Reader)
+	reader.Reset(state.conn)
+	defer func() {
+		reader.Reset(nil)
+		s.readerPool.Put(reader)
+	}()
+	firstRequest := true
+	for {
+		readTimeout := s.ReadHeaderTimeout
+		if readTimeout <= 0 {
+			readTimeout = defaultReadHeaderTimeout
+		}
+		if !firstRequest && s.IdleTimeout > 0 {
+			readTimeout = s.IdleTimeout
+		} else if s.IdleTimeout > 0 && s.IdleTimeout < readTimeout {
+			readTimeout = s.IdleTimeout
+		}
+		_ = state.conn.SetReadDeadline(time.Now().Add(readTimeout))
+		request, closeAfter, err := readRequest(reader, state.conn, s.MaxHeaderBytes, s.MaxBodyBytes, func() {
+			if s.ReadTimeout > 0 {
+				_ = state.conn.SetReadDeadline(time.Now().Add(s.ReadTimeout))
+			} else {
+				_ = state.conn.SetReadDeadline(time.Time{})
+			}
+		})
+		if err != nil {
+			if !errors.Is(err, io.EOF) && !errors.Is(err, net.ErrClosed) && !isTimeout(err) {
+				status := http.StatusBadRequest
+				message := "bad request\n"
+				var protocolErr *protocolError
+				if errors.As(err, &protocolErr) {
+					status = protocolErr.status
+					message = protocolErr.message + "\n"
+				}
+				_ = writeSimple(state.conn, status, message, true, s.WriteTimeout)
+			}
+			return
+		}
+		_ = state.conn.SetReadDeadline(time.Time{})
+		firstRequest = false
+		requestContext, cancel := context.WithCancel(baseCtx)
+		request = request.WithContext(requestContext)
+		state.active.Store(true)
+		writer := s.writerPool.Get().(*responseWriter)
+		writer.reset(request.Method)
+		requestSlot := false
+		if slots := s.requestSlots; slots != nil {
+			select {
+			case slots <- struct{}{}:
+				requestSlot = true
+			case <-baseCtx.Done():
+				cancel()
+				writer.reset("")
+				s.writerPool.Put(writer)
+				state.active.Store(false)
+				return
+			default:
+				writer.Header().Set("Content-Type", "text/plain; charset=utf-8")
+				writer.WriteHeader(http.StatusServiceUnavailable)
+				writer.body.WriteString("server busy\n")
+				closing := s.isClosing() || state.closeWhenIdle.Load()
+				_ = writer.writeTo(state.conn, closeAfter || closing, s.WriteTimeout)
+				cancel()
+				writer.reset("")
+				s.writerPool.Put(writer)
+				state.active.Store(false)
+				if closeAfter || closing {
+					return
+				}
+				continue
+			}
+		}
+		handler := s.handler
+		handler.ServeHTTP(writer, request)
+		cancel()
+
+		closing := s.isClosing() || state.closeWhenIdle.Load()
+		if err := writer.writeTo(state.conn, closeAfter || closing, s.WriteTimeout); err != nil {
+			if requestSlot {
+				<-s.requestSlots
+			}
+			writer.reset("")
+			s.writerPool.Put(writer)
+			state.active.Store(false)
+			return
+		}
+		if requestSlot {
+			<-s.requestSlots
+		}
+		writer.reset("")
+		s.writerPool.Put(writer)
+		state.active.Store(false)
+		if closeAfter || closing {
+			return
+		}
+	}
+}
+
+func (c *connection) close() {
+	if c.closed.CompareAndSwap(false, true) {
+		defer close(c.closeDone)
+		_ = c.conn.Close()
+		c.server.mu.Lock()
+		delete(c.server.conns, c)
+		c.server.mu.Unlock()
+	}
+}
+
+func (s *Server) recycleConnection(c *connection) {
+	<-c.closeDone
+	if c.slotHeld {
+		if slots := s.connSlots; slots != nil {
+			<-slots
+		}
+		c.slotHeld = false
+	}
+	s.connectionPool.Put(c)
+}
+
+type protocolError struct {
+	status  int
+	message string
+}
+
+func (e *protocolError) Error() string { return e.message }
+
+func readRequest(reader *bufio.Reader, conn net.Conn, maxHeaderBytes int, maxBodyBytes int64, setBodyDeadline func()) (*http.Request, bool, error) {
+	if maxHeaderBytes <= 0 {
+		maxHeaderBytes = defaultMaxHeaderBytes
+	}
+	if maxBodyBytes <= 0 {
+		maxBodyBytes = defaultMaxBodyBytes
+	}
+	used := 0
+	readLine := func() (string, error) {
+		line, err := reader.ReadString('\n')
+		used += len(line)
+		if used > maxHeaderBytes {
+			return "", &protocolError{status: http.StatusRequestHeaderFieldsTooLarge, message: "request headers too large"}
+		}
+		if err != nil {
+			return "", err
+		}
+		if !strings.HasSuffix(line, "\n") {
+			return "", &protocolError{status: http.StatusBadRequest, message: "malformed request line"}
+		}
+		return strings.TrimSuffix(strings.TrimSuffix(line, "\n"), "\r"), nil
+	}
+
+	line, err := readLine()
+	if err != nil {
+		return nil, false, err
+	}
+	parts := strings.Split(line, " ")
+	if len(parts) != 3 || !validToken(parts[0]) {
+		return nil, true, &protocolError{status: http.StatusBadRequest, message: "malformed request line"}
+	}
+	protoMajor, protoMinor, ok := parseProtocol(parts[2])
+	if !ok {
+		return nil, true, &protocolError{status: http.StatusHTTPVersionNotSupported, message: "unsupported HTTP version"}
+	}
+	target := parts[1]
+	var requestURL *url.URL
+	if target == "*" {
+		requestURL = &url.URL{Path: "*"}
+	} else {
+		requestURL, err = url.ParseRequestURI(target)
+		if err != nil || requestURL.Path == "" || !strings.HasPrefix(requestURL.Path, "/") {
+			return nil, true, &protocolError{status: http.StatusBadRequest, message: "malformed request target"}
+		}
+	}
+
+	header := make(http.Header)
+	for {
+		line, err = readLine()
+		if err != nil {
+			return nil, true, err
+		}
+		if line == "" {
+			break
+		}
+		if strings.HasPrefix(line, " ") || strings.HasPrefix(line, "\t") {
+			return nil, true, &protocolError{status: http.StatusBadRequest, message: "folded headers are not supported"}
+		}
+		index := strings.IndexByte(line, ':')
+		if index <= 0 {
+			return nil, true, &protocolError{status: http.StatusBadRequest, message: "malformed header"}
+		}
+		name := strings.TrimSpace(line[:index])
+		value := strings.TrimSpace(line[index+1:])
+		if !validToken(name) || strings.ContainsAny(value, "\r\n") {
+			return nil, true, &protocolError{status: http.StatusBadRequest, message: "malformed header"}
+		}
+		header.Add(name, value)
+	}
+
+	if value := header.Get("Expect"); value != "" {
+		return nil, true, &protocolError{status: http.StatusExpectationFailed, message: "unsupported expectation"}
+	}
+	if value := header.Get("Transfer-Encoding"); value != "" && !strings.EqualFold(value, "identity") {
+		return nil, true, &protocolError{status: http.StatusNotImplemented, message: "chunked requests are not supported"}
+	}
+	contentLength := int64(0)
+	if value := header.Get("Content-Length"); value != "" {
+		contentLength, err = strconv.ParseInt(strings.TrimSpace(value), 10, 64)
+		if err != nil || contentLength < 0 {
+			return nil, true, &protocolError{status: http.StatusBadRequest, message: "invalid content length"}
+		}
+	}
+	if contentLength > maxBodyBytes {
+		return nil, true, &protocolError{status: http.StatusRequestEntityTooLarge, message: "request body too large"}
+	}
+	body := make([]byte, contentLength)
+	if contentLength > 0 {
+		if setBodyDeadline != nil {
+			setBodyDeadline()
+		}
+		if _, err := io.ReadFull(reader, body); err != nil {
+			return nil, true, err
+		}
+	}
+	request := &http.Request{
+		Method:        parts[0],
+		URL:           requestURL,
+		Proto:         parts[2],
+		ProtoMajor:    protoMajor,
+		ProtoMinor:    protoMinor,
+		Header:        header,
+		Body:          io.NopCloser(bytes.NewReader(body)),
+		ContentLength: contentLength,
+		Host:          header.Get("Host"),
+		RequestURI:    target,
+		RemoteAddr:    conn.RemoteAddr().String(),
+	}
+	closeAfter := strings.EqualFold(header.Get("Connection"), "close") ||
+		(protoMajor == 1 && protoMinor == 0 && !strings.Contains(strings.ToLower(header.Get("Connection")), "keep-alive"))
+	request.Close = closeAfter
+	return request, closeAfter, nil
+}
+
+func parseProtocol(value string) (int, int, bool) {
+	if value != "HTTP/1.1" && value != "HTTP/1.0" {
+		return 0, 0, false
+	}
+	return 1, int(value[len(value)-1] - '0'), true
+}
+
+func validToken(value string) bool {
+	if value == "" {
+		return false
+	}
+	for _, char := range value {
+		if char <= ' ' || strings.ContainsRune("()<>@,;:\\\"/[]?={} \t", char) {
+			return false
+		}
+	}
+	return true
+}
+
+func isTimeout(err error) bool {
+	var netErr net.Error
+	return errors.As(err, &netErr) && netErr.Timeout()
+}
+
+func writeSimple(conn net.Conn, status int, body string, closeAfter bool, timeout time.Duration) error {
+	writer := newResponseWriter("")
+	writer.status = status
+	writer.wroteHeader = true
+	writer.body.WriteString(body)
+	writer.header.Set("Content-Type", "text/plain; charset=utf-8")
+	return writer.writeTo(conn, closeAfter, timeout)
+}
+
+type responseWriter struct {
+	header        http.Header
+	status        int
+	wroteHeader   bool
+	body          bytes.Buffer
+	requestMethod string
+}
+
+func newResponseWriter(method string) *responseWriter {
+	return &responseWriter{header: make(http.Header), requestMethod: method}
+}
+
+func (w *responseWriter) reset(method string) {
+	for key := range w.header {
+		delete(w.header, key)
+	}
+	w.status = 0
+	w.wroteHeader = false
+	if w.body.Cap() > maxPooledResponseBytes {
+		w.body = bytes.Buffer{}
+	} else {
+		w.body.Reset()
+	}
+	w.requestMethod = method
+}
+
+func (w *responseWriter) Header() http.Header { return w.header }
+
+func (w *responseWriter) WriteHeader(status int) {
+	if w.wroteHeader {
+		return
+	}
+	w.status = status
+	w.wroteHeader = true
+}
+
+func (w *responseWriter) Write(data []byte) (int, error) {
+	if !w.wroteHeader {
+		w.WriteHeader(http.StatusOK)
+	}
+	return w.body.Write(data)
+}
+
+func (w *responseWriter) writeTo(conn net.Conn, closeAfter bool, timeout time.Duration) error {
+	if timeout > 0 {
+		_ = conn.SetWriteDeadline(time.Now().Add(timeout))
+	}
+	if !w.wroteHeader {
+		w.status = http.StatusOK
+		w.wroteHeader = true
+	}
+	body := w.body.Bytes()
+	noBody := w.requestMethod == http.MethodHead || w.status == http.StatusNoContent ||
+		w.status == http.StatusNotModified || (w.status >= 100 && w.status < 200)
+	if noBody {
+		body = nil
+	}
+	if w.header.Get("Content-Length") == "" && w.status != http.StatusNoContent && w.status != http.StatusNotModified {
+		w.header.Set("Content-Length", strconv.Itoa(len(w.body.Bytes())))
+	}
+	if closeAfter {
+		w.header.Set("Connection", "close")
+	}
+	keys := make([]string, 0, len(w.header))
+	for key := range w.header {
+		keys = append(keys, key)
+	}
+	sort.Strings(keys)
+	var output bytes.Buffer
+	output.WriteString("HTTP/1.1 ")
+	output.WriteString(strconv.Itoa(w.status))
+	output.WriteByte(' ')
+	output.WriteString(http.StatusText(w.status))
+	output.WriteString("\r\n")
+	for _, key := range keys {
+		for _, value := range w.header.Values(key) {
+			if strings.ContainsAny(value, "\r\n") {
+				return errors.New("invalid response header")
+			}
+			output.WriteString(key)
+			output.WriteString(": ")
+			output.WriteString(value)
+			output.WriteString("\r\n")
+		}
+	}
+	output.WriteString("\r\n")
+	output.Write(body)
+	_, err := conn.Write(output.Bytes())
+	_ = conn.SetWriteDeadline(time.Time{})
+	return err
+}

transport/native/server_test.go

@@ -0,0 +1,301 @@
+package native_test
+
+import (
+	"bufio"
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"strings"
+	"testing"
+	"time"
+
+	omni "git.yuntuo.hk/k2safe/go-omni"
+	"git.yuntuo.hk/k2safe/go-omni/transport/native"
+	netrouter "git.yuntuo.hk/k2safe/go-omni/transport/nethttp"
+)
+
+func TestServerKeepAliveAndApplicationBridge(t *testing.T) {
+	application := omni.NewWithRouter(netrouter.New())
+	application.GET("/hello", func(ctx *omni.Context) error {
+		return ctx.OK(map[string]string{"message": "hello"})
+	})
+	server, listener, done := startServer(t, application)
+	defer stopServer(t, server, listener, done)
+
+	conn, err := net.Dial("tcp", listener.Addr().String())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+	reader := bufio.NewReader(conn)
+
+	for index, closeAfter := range []bool{false, true} {
+		connectionHeader := "keep-alive"
+		if closeAfter {
+			connectionHeader = "close"
+		}
+		fmt.Fprintf(conn, "GET /hello HTTP/1.1\r\nHost: test\r\nConnection: %s\r\n\r\n", connectionHeader)
+		request := &http.Request{Method: http.MethodGet}
+		response, err := http.ReadResponse(reader, request)
+		if err != nil {
+			t.Fatalf("response %d: %v", index, err)
+		}
+		body, err := io.ReadAll(response.Body)
+		response.Body.Close()
+		if err != nil || response.StatusCode != http.StatusOK || !strings.Contains(string(body), "\"message\":\"hello\"") {
+			t.Fatalf("response %d: status=%d body=%s error=%v", index, response.StatusCode, body, err)
+		}
+	}
+}
+
+func TestServerRejectsUnsupportedBodyAndLimits(t *testing.T) {
+	application := omni.NewWithRouter(netrouter.New())
+	application.POST("/body", func(ctx *omni.Context) error { return ctx.OK("ok") })
+
+	server := native.New(application)
+	server.MaxBodyBytes = 4
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	done := make(chan error, 1)
+	go func() { done <- server.Serve(context.Background(), listener) }()
+	waitForServer(t, listener.Addr().String())
+	defer stopServer(t, server, listener, done)
+
+	for _, requestText := range []string{
+		"POST /body HTTP/1.1\r\nHost: test\r\nTransfer-Encoding: chunked\r\n\r\n4\r\ntest\r\n0\r\n\r\n",
+		"POST /body HTTP/1.1\r\nHost: test\r\nContent-Length: 5\r\n\r\ntest!",
+	} {
+		conn, err := net.Dial("tcp", listener.Addr().String())
+		if err != nil {
+			t.Fatal(err)
+		}
+		_, _ = io.WriteString(conn, requestText)
+		response, err := http.ReadResponse(bufio.NewReader(conn), &http.Request{Method: http.MethodPost})
+		if err != nil {
+			conn.Close()
+			t.Fatal(err)
+		}
+		response.Body.Close()
+		if response.StatusCode != http.StatusNotImplemented && response.StatusCode != http.StatusRequestEntityTooLarge {
+			t.Fatalf("unexpected protocol status %d", response.StatusCode)
+		}
+		conn.Close()
+	}
+}
+
+func TestServerShutdownCancelsActiveHandler(t *testing.T) {
+	started := make(chan struct{})
+	application := omni.NewWithRouter(netrouter.New())
+	application.GET("/wait", func(ctx *omni.Context) error {
+		close(started)
+		<-ctx.Context().Done()
+		return ctx.Context().Err()
+	})
+	server, listener, done := startServer(t, application)
+
+	conn, err := net.Dial("tcp", listener.Addr().String())
+	if err != nil {
+		t.Fatal(err)
+	}
+	_, _ = io.WriteString(conn, "GET /wait HTTP/1.1\r\nHost: test\r\nConnection: close\r\n\r\n")
+	select {
+	case <-started:
+	case <-time.After(2 * time.Second):
+		t.Fatal("handler did not start")
+	}
+	shutdownContext, cancel := context.WithTimeout(context.Background(), time.Second)
+	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)
+	}
+	conn.Close()
+	select {
+	case err := <-done:
+		if err != nil && !errors.Is(err, net.ErrClosed) {
+			t.Fatal(err)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("server did not stop")
+	}
+}
+
+func TestServerConcurrentRequestLimit(t *testing.T) {
+	started := make(chan struct{})
+	release := make(chan struct{})
+	application := omni.NewWithRouter(netrouter.New())
+	application.GET("/work", func(ctx *omni.Context) error {
+		close(started)
+		<-release
+		return ctx.OK("done")
+	})
+	server := native.New(application)
+	server.MaxConcurrentRequests = 1
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	done := make(chan error, 1)
+	go func() { done <- server.Serve(context.Background(), listener) }()
+	waitForServer(t, listener.Addr().String())
+	defer func() {
+		stopServer(t, server, listener, done)
+	}()
+
+	first, err := net.Dial("tcp", listener.Addr().String())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer first.Close()
+	_, _ = io.WriteString(first, "GET /work HTTP/1.1\r\nHost: test\r\nConnection: close\r\n\r\n")
+	select {
+	case <-started:
+	case <-time.After(2 * time.Second):
+		t.Fatal("first handler did not start")
+	}
+
+	second, err := net.Dial("tcp", listener.Addr().String())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer second.Close()
+	_, _ = io.WriteString(second, "GET /work HTTP/1.1\r\nHost: test\r\nConnection: close\r\n\r\n")
+	response, err := http.ReadResponse(bufio.NewReader(second), &http.Request{Method: http.MethodGet})
+	if err != nil {
+		t.Fatal(err)
+	}
+	response.Body.Close()
+	if response.StatusCode != http.StatusServiceUnavailable {
+		t.Fatalf("expected 503 at concurrency limit, got %d", response.StatusCode)
+	}
+	close(release)
+	response, err = http.ReadResponse(bufio.NewReader(first), &http.Request{Method: http.MethodGet})
+	if err != nil {
+		t.Fatal(err)
+	}
+	response.Body.Close()
+	if response.StatusCode != http.StatusOK {
+		t.Fatalf("expected first request to complete, got %d", response.StatusCode)
+	}
+}
+
+func TestServerReadHeaderTimeout(t *testing.T) {
+	application := omni.NewWithRouter(netrouter.New())
+	application.GET("/health", func(ctx *omni.Context) error { return ctx.OK("ok") })
+	server := native.New(application)
+	server.ReadHeaderTimeout = 25 * time.Millisecond
+	server.IdleTimeout = 0
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	done := make(chan error, 1)
+	go func() { done <- server.Serve(context.Background(), listener) }()
+	waitForServer(t, listener.Addr().String())
+	defer stopServer(t, server, listener, done)
+
+	conn, err := net.Dial("tcp", listener.Addr().String())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+	conn.SetReadDeadline(time.Now().Add(time.Second))
+	_, _ = io.WriteString(conn, "GET /health HTTP/1.1\r\nHost: test")
+	buffer := make([]byte, 1)
+	if _, err := conn.Read(buffer); err == nil {
+		t.Fatal("expected incomplete headers to time out and close")
+	}
+}
+
+func TestAppRunWithNativeRunner(t *testing.T) {
+	application := omni.NewWithRouter(netrouter.New())
+	application.GET("/health", func(ctx *omni.Context) error { return ctx.OK("ok") })
+	address := freeAddress(t)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	done := make(chan error, 1)
+	go func() { done <- application.RunWith(ctx, native.New(nil), address) }()
+	waitForServer(t, address)
+	response, err := http.Get("http://" + address + "/health")
+	if err != nil {
+		t.Fatal(err)
+	}
+	response.Body.Close()
+	if response.StatusCode != http.StatusOK {
+		t.Fatalf("unexpected status %d", response.StatusCode)
+	}
+	cancel()
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Fatal(err)
+		}
+	case <-time.After(3 * time.Second):
+		t.Fatal("RunWith did not stop")
+	}
+}
+
+func startServer(t *testing.T, handler http.Handler) (*native.Server, net.Listener, <-chan error) {
+	t.Helper()
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	server := native.New(handler)
+	done := make(chan error, 1)
+	go func() { done <- server.Serve(context.Background(), listener) }()
+	waitForServer(t, listener.Addr().String())
+	return server, listener, done
+}
+
+func stopServer(t *testing.T, server *native.Server, listener net.Listener, done <-chan error) {
+	t.Helper()
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	_ = server.Shutdown(ctx)
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("server did not stop")
+	}
+	_ = listener.Close()
+}
+
+func freeAddress(t *testing.T) string {
+	t.Helper()
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	address := listener.Addr().String()
+	listener.Close()
+	return address
+}
+
+func waitForServer(t *testing.T, address string) {
+	t.Helper()
+	deadline := time.Now().Add(2 * time.Second)
+	for {
+		conn, err := net.DialTimeout("tcp", address, 50*time.Millisecond)
+		if err == nil {
+			conn.Close()
+			return
+		}
+		if time.Now().After(deadline) {
+			t.Fatalf("server did not start: %v", err)
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+}