feat: add opt-in HTTP2 transport experiment

This commit is contained in:
2026-09-22 19:08:19 +08:00
parent 83633f724e
commit fcd6f48b53
7 changed files with 621 additions and 1 deletions

View File

@@ -14,6 +14,8 @@ registerRoutes(application) // 与 omni.New() 复用同一份业务注册代码
`transport/nethttp` 是当前内置的标准库路由实现。它提供一套可移植的路径、方法和错误回退规则;自研网络层可以直接实现同一个 `httpx.Router` 契约。
`transport/http2` 是独立的 opt-in 实验 Runner,负责连接接入、TLS/ALPN、连接上限和全局请求背压,底层帧解析与 stream 流控暂由 `golang.org/x/net/http2` 提供。它不替换 `transport/native`,也不改变默认启动器;只有通过 `App.RunWith(ctx, http2.New(nil), address)` 显式启用才会运行。
`transport/native` 是第一版自研 HTTP/1.1 服务端:它自己管理 TCP listener、连接、Keep-Alive、请求解析、响应写回和优雅关闭,再把请求交给现有 `http.Handler`。请求体支持 `Content-Length` 和 chunked,并统一受 `MaxBodyBytes` 限制;其他传输编码会被拒绝。它仍使用标准 `http.Request` / `http.ResponseWriter` 作为业务桥接,因此不会影响现有 Handler / Service。
native 提供 `server.Stats()` 读取已接受连接数、连接拒绝数、当前连接数和当前请求数,适合和 `metrics.Metrics` 的业务请求指标一起采集;读取不会阻塞连接生命周期。`RejectedConnections` 统计连接闸门拒绝,`RejectedRequests` 统计并发请求闸门返回 503。

383
transport/http2/server.go Normal file
View File

@@ -0,0 +1,383 @@
// Package http2 provides an opt-in HTTP/2 transport experiment.
//
// It owns the listener, TLS/ALPN setup, connection admission and global
// request backpressure while using x/net/http2 for frame parsing and stream
// flow control. It is deliberately separate from transport/native, whose
// production preview remains HTTP/1.1.
package http2
import (
"context"
"crypto/tls"
"errors"
"fmt"
"net"
"net/http"
"sync"
"sync/atomic"
"time"
xhttp2 "golang.org/x/net/http2"
)
const (
defaultShutdownTimeout = 10 * time.Second
defaultTLSHandshake = 10 * time.Second
http2ALPNProtocol = "h2"
)
// Server is an opt-in HTTP/2 transport. Run uses cleartext h2c for local
// experiments; RunTLS or ServeTLS should be used for deployed TLS traffic.
// HTTP/2 stream limits and global request admission are independent: a client
// may open many streams, but only admitted handlers run concurrently.
type Server struct {
MaxConnections int
MaxConcurrentRequests int
MaxConcurrentStreams uint32
IdleTimeout time.Duration
TLSHandshakeTimeout time.Duration
ShutdownTimeout time.Duration
TLSConfig *tls.Config
mu sync.Mutex
handler http.Handler
listener net.Listener
closing bool
cancel context.CancelFunc
conns map[net.Conn]struct{}
connSlots chan struct{}
requestSlots chan struct{}
accepted atomic.Uint64
rejectedConns atomic.Uint64
rejectedReqs atomic.Uint64
activeConns atomic.Int64
activeRequests atomic.Int64
wg sync.WaitGroup
}
// Stats is a point-in-time view of HTTP/2 transport admission and lifecycle.
// Accepted and rejected counters are monotonic; active counters are gauges.
type Stats struct {
AcceptedConnections uint64 `json:"acceptedConnections"`
RejectedConnections uint64 `json:"rejectedConnections"`
RejectedRequests uint64 `json:"rejectedRequests"`
ActiveConnections int64 `json:"activeConnections"`
ActiveRequests int64 `json:"activeRequests"`
}
// New creates an HTTP/2 transport with the supplied handler. The handler may
// also be supplied by App.RunWith through the Runner interface.
func New(handler http.Handler) *Server {
return &Server{handler: handler, conns: make(map[net.Conn]struct{})}
}
// Stats returns a lock-free snapshot suitable for a metrics handler.
func (s *Server) Stats() Stats {
if s == nil {
return Stats{}
}
return Stats{
AcceptedConnections: s.accepted.Load(),
RejectedConnections: s.rejectedConns.Load(),
RejectedRequests: s.rejectedReqs.Load(),
ActiveConnections: s.activeConns.Load(),
ActiveRequests: s.activeRequests.Load(),
}
}
// Run implements app.Runner. Without TLSConfig it starts an h2c listener for
// local testing. Use RunTLS or TLSConfig for real HTTPS traffic.
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/http2: nil handler")
}
if address == "" {
address = ":8080"
}
listener, err := net.Listen("tcp", address)
if err != nil {
return err
}
if s.TLSConfig != nil {
return s.ServeTLS(ctx, listener, s.TLSConfig)
}
return s.Serve(ctx, listener)
}
// RunTLS loads a certificate pair and serves HTTP/2 over TLS with ALPN h2.
func (s *Server) RunTLS(ctx context.Context, handler http.Handler, address, certFile, keyFile string) error {
if handler != nil {
s.handler = handler
}
if s.handler == nil {
return errors.New("go-omni/http2: nil handler")
}
if address == "" {
address = ":8443"
}
certificate, err := tls.LoadX509KeyPair(certFile, keyFile)
if err != nil {
return err
}
listener, err := net.Listen("tcp", address)
if err != nil {
return err
}
return s.ServeTLS(ctx, listener, &tls.Config{Certificates: []tls.Certificate{certificate}, MinVersion: tls.VersionTLS12})
}
// ServeTLS wraps listener with TLS and requires ALPN h2. A caller-provided
// config is cloned, so the original value is never mutated.
func (s *Server) ServeTLS(ctx context.Context, listener net.Listener, config *tls.Config) error {
if listener == nil {
return errors.New("go-omni/http2: nil listener")
}
if config == nil {
return errors.New("go-omni/http2: nil TLS config")
}
prepared, err := prepareTLSConfig(config)
if err != nil {
_ = listener.Close()
return err
}
return s.Serve(ctx, tls.NewListener(listener, prepared))
}
func prepareTLSConfig(config *tls.Config) (*tls.Config, error) {
clone := config.Clone()
if len(clone.NextProtos) > 0 {
for _, protocol := range clone.NextProtos {
if protocol != http2ALPNProtocol {
return nil, fmt.Errorf("go-omni/http2: unsupported TLS ALPN protocol %q; only %q is supported", protocol, http2ALPNProtocol)
}
}
}
clone.NextProtos = []string{http2ALPNProtocol}
return clone, nil
}
// Serve accepts h2c connections from listener until ctx is canceled.
func (s *Server) Serve(ctx context.Context, listener net.Listener) error {
if listener == nil {
return errors.New("go-omni/http2: nil listener")
}
if s.handler == nil {
return errors.New("go-omni/http2: nil handler")
}
if ctx == nil {
ctx = context.Background()
}
s.mu.Lock()
if s.cancel != nil {
s.mu.Unlock()
return errors.New("go-omni/http2: server is already running")
}
baseCtx, cancel := context.WithCancel(ctx)
s.cancel = cancel
s.listener = listener
s.closing = false
if s.conns == nil {
s.conns = make(map[net.Conn]struct{})
}
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
}
s.mu.Unlock()
gate := &requestGate{server: s, next: s.handler, slots: s.requestSlots}
protocolServer := &xhttp2.Server{
MaxConcurrentStreams: s.MaxConcurrentStreams,
IdleTimeout: s.IdleTimeout,
}
shutdownDone := make(chan struct{})
go func() {
select {
case <-ctx.Done():
shutdownCtx, cancel := context.WithTimeout(context.Background(), s.shutdownTimeout())
_ = s.Shutdown(shutdownCtx)
cancel()
case <-shutdownDone:
}
}()
defer func() {
close(shutdownDone)
cancel()
s.mu.Lock()
s.listener = nil
s.cancel = nil
s.mu.Unlock()
}()
for {
conn, err := listener.Accept()
if err != nil {
if ctx.Err() != nil || s.isClosing() {
break
}
return err
}
s.accepted.Add(1)
if slots := s.connSlots; slots != nil {
select {
case slots <- struct{}{}:
default:
s.rejectedConns.Add(1)
_ = conn.Close()
continue
}
}
s.mu.Lock()
if s.closing {
s.mu.Unlock()
if s.connSlots != nil {
<-s.connSlots
}
_ = conn.Close()
continue
}
s.conns[conn] = struct{}{}
s.activeConns.Add(1)
s.wg.Add(1)
s.mu.Unlock()
go s.serveConn(baseCtx, protocolServer, gate, conn)
}
return nil
}
func (s *Server) shutdownTimeout() time.Duration {
if s.ShutdownTimeout > 0 {
return s.ShutdownTimeout
}
return defaultShutdownTimeout
}
func (s *Server) serveConn(ctx context.Context, protocolServer *xhttp2.Server, handler http.Handler, conn net.Conn) {
defer s.wg.Done()
defer func() {
s.mu.Lock()
delete(s.conns, conn)
s.activeConns.Add(-1)
s.mu.Unlock()
if s.connSlots != nil {
<-s.connSlots
}
_ = conn.Close()
}()
if tlsConn, ok := conn.(*tls.Conn); ok {
_ = conn.SetDeadline(time.Now().Add(s.tlsHandshakeTimeout()))
if err := tlsConn.HandshakeContext(ctx); err != nil {
return
}
_ = conn.SetDeadline(time.Time{})
}
protocolServer.ServeConn(conn, &xhttp2.ServeConnOpts{
Context: ctx,
Handler: handler,
BaseConfig: &http.Server{
IdleTimeout: s.IdleTimeout,
},
})
}
func (s *Server) tlsHandshakeTimeout() time.Duration {
if s.TLSHandshakeTimeout > 0 {
return s.TLSHandshakeTimeout
}
return defaultTLSHandshake
}
// Shutdown stops accepting connections and waits for active HTTP/2 sessions.
// If the context expires, remaining connections are closed so handlers can
// observe cancellation through their request contexts.
func (s *Server) Shutdown(ctx context.Context) error {
if s == nil {
return nil
}
if ctx == nil {
timeout := s.ShutdownTimeout
if timeout <= 0 {
timeout = defaultShutdownTimeout
}
var cancel context.CancelFunc
ctx, cancel = context.WithTimeout(context.Background(), timeout)
defer cancel()
}
s.mu.Lock()
if s.cancel == nil {
s.mu.Unlock()
return nil
}
s.closing = true
listener := s.listener
cancel := s.cancel
connections := make([]net.Conn, 0, len(s.conns))
for conn := range s.conns {
connections = append(connections, conn)
}
s.mu.Unlock()
if listener != nil {
_ = listener.Close()
}
cancel()
done := make(chan struct{})
go func() {
s.wg.Wait()
close(done)
}()
select {
case <-done:
return nil
case <-ctx.Done():
for _, conn := range connections {
_ = conn.Close()
}
return ctx.Err()
}
}
func (s *Server) isClosing() bool {
s.mu.Lock()
closing := s.closing
s.mu.Unlock()
return closing
}
type requestGate struct {
server *Server
next http.Handler
slots chan struct{}
}
func (g *requestGate) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
if g.slots != nil {
select {
case g.slots <- struct{}{}:
g.server.activeRequests.Add(1)
defer func() {
<-g.slots
g.server.activeRequests.Add(-1)
}()
default:
g.server.rejectedReqs.Add(1)
http.Error(writer, "server busy\n", http.StatusServiceUnavailable)
return
}
} else {
g.server.activeRequests.Add(1)
defer g.server.activeRequests.Add(-1)
}
g.next.ServeHTTP(writer, request)
}

View File

@@ -0,0 +1,230 @@
package http2
import (
"context"
"crypto/rand"
"crypto/rsa"
"crypto/tls"
"crypto/x509"
"crypto/x509/pkix"
"errors"
"math/big"
"net"
"net/http"
"sync"
"testing"
"time"
xhttp2 "golang.org/x/net/http2"
)
func TestServerHandlesH2CRequest(t *testing.T) {
server := New(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
if request.ProtoMajor != 2 {
http.Error(writer, "not HTTP/2", http.StatusInternalServerError)
return
}
_, _ = writer.Write([]byte("ok"))
}))
address, stop := startServer(t, server)
defer stop()
transport := &xhttp2.Transport{
AllowHTTP: true,
DialTLSContext: func(ctx context.Context, network, address string, _ *tls.Config) (net.Conn, error) {
return (&net.Dialer{}).DialContext(ctx, network, address)
},
}
client := &http.Client{Transport: transport, Timeout: 2 * time.Second}
defer transport.CloseIdleConnections()
response, err := client.Get("http://" + address + "/health")
if err != nil {
t.Fatal(err)
}
defer response.Body.Close()
if response.StatusCode != http.StatusOK {
t.Fatalf("unexpected status: %d", response.StatusCode)
}
if stats := server.Stats(); stats.AcceptedConnections != 1 {
t.Fatalf("unexpected HTTP/2 stats: %+v", stats)
}
}
func TestServerEnforcesGlobalRequestLimitAndRecovers(t *testing.T) {
started := make(chan struct{})
release := make(chan struct{})
var once sync.Once
server := New(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
if request.URL.Path == "/hold" {
once.Do(func() { close(started) })
<-release
}
_, _ = writer.Write([]byte("ok"))
}))
server.MaxConcurrentRequests = 1
address, stop := startServer(t, server)
defer stop()
transport := &xhttp2.Transport{
AllowHTTP: true,
DialTLSContext: func(ctx context.Context, network, address string, _ *tls.Config) (net.Conn, error) {
return (&net.Dialer{}).DialContext(ctx, network, address)
},
}
client := &http.Client{Transport: transport, Timeout: 3 * time.Second}
defer transport.CloseIdleConnections()
firstDone := make(chan *http.Response, 1)
firstErr := make(chan error, 1)
go func() {
response, err := client.Get("http://" + address + "/hold")
if err != nil {
firstErr <- err
return
}
firstDone <- response
}()
select {
case <-started:
case err := <-firstErr:
t.Fatal(err)
case <-time.After(time.Second):
t.Fatal("first HTTP/2 stream did not start")
}
response, err := client.Get("http://" + address + "/busy")
if err != nil {
t.Fatal(err)
}
response.Body.Close()
if response.StatusCode != http.StatusServiceUnavailable {
t.Fatalf("expected 503 at stream limit, got %d", response.StatusCode)
}
if stats := server.Stats(); stats.RejectedRequests != 1 {
t.Fatalf("expected one rejected stream, got %+v", stats)
}
close(release)
select {
case response = <-firstDone:
response.Body.Close()
case err := <-firstErr:
t.Fatal(err)
case <-time.After(time.Second):
t.Fatal("held HTTP/2 stream did not finish")
}
response, err = client.Get("http://" + address + "/recovered")
if err != nil {
t.Fatal(err)
}
response.Body.Close()
if response.StatusCode != http.StatusOK {
t.Fatalf("request limit did not recover, got %d", response.StatusCode)
}
}
func TestServerTLSUsesH2ALPN(t *testing.T) {
certificate, roots := testCertificate(t)
server := New(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
if request.TLS == nil || request.TLS.NegotiatedProtocol != "h2" {
http.Error(writer, "missing h2 ALPN", http.StatusInternalServerError)
return
}
writer.WriteHeader(http.StatusNoContent)
}))
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.ServeTLS(context.Background(), listener, &tls.Config{
Certificates: []tls.Certificate{certificate}, MinVersion: tls.VersionTLS12,
})
}()
stop := func() {
t.Helper()
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
if err := server.Shutdown(ctx); err != nil && !errors.Is(err, context.DeadlineExceeded) {
t.Errorf("shutdown: %v", err)
}
if err := <-done; err != nil {
t.Errorf("serve: %v", err)
}
}
defer stop()
transport := &xhttp2.Transport{TLSClientConfig: &tls.Config{
RootCAs: roots, ServerName: "localhost", MinVersion: tls.VersionTLS12,
NextProtos: []string{"h2"},
}}
client := &http.Client{Transport: transport, Timeout: 2 * time.Second}
defer transport.CloseIdleConnections()
response, err := client.Get("https://" + listener.Addr().String() + "/health")
if err != nil {
t.Fatal(err)
}
response.Body.Close()
if response.StatusCode != http.StatusNoContent {
t.Fatalf("unexpected TLS status: %d", response.StatusCode)
}
}
func startServer(t *testing.T, server *Server) (string, func()) {
t.Helper()
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) }()
stop := func() {
t.Helper()
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
if err := server.Shutdown(ctx); err != nil && !errors.Is(err, context.DeadlineExceeded) {
t.Errorf("shutdown: %v", err)
}
if err := <-done; err != nil {
t.Errorf("serve: %v", err)
}
}
return listener.Addr().String(), stop
}
func testCertificate(t *testing.T) (tls.Certificate, *x509.CertPool) {
t.Helper()
key, err := rsa.GenerateKey(rand.Reader, 2048)
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)
}
template := &x509.Certificate{
SerialNumber: serial,
Subject: pkix.Name{CommonName: "localhost"},
DNSNames: []string{"localhost"},
NotBefore: time.Now().Add(-time.Minute),
NotAfter: time.Now().Add(time.Hour),
KeyUsage: x509.KeyUsageKeyEncipherment | x509.KeyUsageDigitalSignature,
ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
}
der, err := x509.CreateCertificate(rand.Reader, template, template, &key.PublicKey, key)
if err != nil {
t.Fatal(err)
}
leaf, err := x509.ParseCertificate(der)
if err != nil {
t.Fatal(err)
}
certificate := tls.Certificate{
Certificate: [][]byte{der},
PrivateKey: key,
}
certificate.Leaf = leaf
roots := x509.NewCertPool()
roots.AddCert(leaf)
return certificate, roots
}