feat: observe HTTP2 reset and goaway frames

transport/http2/frameObserver.go

@@ -0,0 +1,106 @@
+package http2
+
+import (
+	"crypto/tls"
+	"net"
+	"sync"
+)
+
+const (
+	frameTypeRSTStream byte = 0x3
+	frameTypeGoAway    byte = 0x7
+)
+
+type frameDirection uint8
+
+const (
+	frameRead frameDirection = iota
+	frameWrite
+)
+
+// observedConn keeps the HTTP/2 implementation opaque while exposing the
+// small amount of frame-level telemetry that x/net/http2 does not provide.
+// It parses only the nine-byte frame header and skips payload bytes without
+// buffering them, so large DATA frames cannot increase telemetry memory.
+type observedConn struct {
+	net.Conn
+	onFrame     func(frameDirection, byte)
+	readParser  frameParser
+	writeParser frameParser
+	writeMu     sync.Mutex
+}
+
+func (c *observedConn) Read(buffer []byte) (int, error) {
+	n, err := c.Conn.Read(buffer)
+	if n > 0 {
+		c.readParser.observe(buffer[:n], frameRead, c.onFrame)
+	}
+	return n, err
+}
+
+func (c *observedConn) Write(buffer []byte) (int, error) {
+	c.writeMu.Lock()
+	defer c.writeMu.Unlock()
+	n, err := c.Conn.Write(buffer)
+	if n > 0 {
+		c.writeParser.observe(buffer[:n], frameWrite, c.onFrame)
+	}
+	return n, err
+}
+
+// observedTLSConn preserves the connection state hook used by x/net/http2 so
+// handlers still receive Request.TLS after frame observation is added.
+type observedTLSConn struct {
+	*observedConn
+	state tls.ConnectionState
+}
+
+func (c *observedTLSConn) ConnectionState() tls.ConnectionState { return c.state }
+
+type frameParser struct {
+	prefaceRemain int
+	header        [9]byte
+	headerBytes   int
+	payloadRemain int
+	frameType     byte
+}
+
+func (p *frameParser) observe(buffer []byte, direction frameDirection, onFrame func(frameDirection, byte)) {
+	for len(buffer) > 0 {
+		if p.prefaceRemain > 0 {
+			skip := p.prefaceRemain
+			if skip > len(buffer) {
+				skip = len(buffer)
+			}
+			p.prefaceRemain -= skip
+			buffer = buffer[skip:]
+			continue
+		}
+		if p.payloadRemain > 0 {
+			skip := p.payloadRemain
+			if skip > len(buffer) {
+				skip = len(buffer)
+			}
+			p.payloadRemain -= skip
+			buffer = buffer[skip:]
+			if p.payloadRemain > 0 {
+				continue
+			}
+			onFrame(direction, p.frameType)
+			continue
+		}
+
+		copied := copy(p.header[p.headerBytes:], buffer)
+		p.headerBytes += copied
+		buffer = buffer[copied:]
+		if p.headerBytes < len(p.header) {
+			continue
+		}
+		p.payloadRemain = int(p.header[0])<<16 | int(p.header[1])<<8 | int(p.header[2])
+		p.frameType = p.header[3]
+		p.headerBytes = 0
+		if p.payloadRemain == 0 {
+			onFrame(direction, p.frameType)
+		}
+	}
+}

transport/http2/lifecycle_test.go

@@ -160,6 +160,9 @@ func TestShutdownSendsGoAwayAndDrains(t *testing.T) {
 	if err := rctx.Err(); err != nil {
 		t.Fatalf("graceful shutdown canceled active request: %v", err)
 	}
+	if stats := s.Stats(); stats.GoAwayFrames != 1 {
+		t.Fatalf("expected one observed GOAWAY frame, got %+v", stats)
+	}
 	select {
 	case err := <-done:
 		t.Fatalf("Serve returned before drain: %v", err)
@@ -214,6 +217,9 @@ func TestRSTStreamCancelsOnlyOneRequest(t *testing.T) {
 	case <-time.After(time.Second):
 		t.Fatal("RST did not cancel request")
 	}
+	if stats := s.Stats(); stats.RSTStreams != 1 {
+		t.Fatalf("expected one observed RST_STREAM, got %+v", stats)
+	}
 	release <- struct{}{}
 	if got := p.body(t, 3); got != "/survive" {
 		t.Fatalf("sibling body=%q", got)

transport/http2/server.go

@@ -59,6 +59,8 @@ type Server struct {
 	rejectedReqs   atomic.Uint64
 	protocolErrors atomic.Uint64
 	goAwaySent     atomic.Uint64
+	rstStreams     atomic.Uint64
+	goAwayFrames   atomic.Uint64
 	activeConns    atomic.Int64
 	activeRequests atomic.Int64
 	wg             sync.WaitGroup
@@ -72,6 +74,8 @@ type Stats struct {
 	RejectedRequests    uint64 `json:"rejectedRequests"`
 	ProtocolErrors      uint64 `json:"protocolErrors"`
 	GoAwaySent          uint64 `json:"goAwaySent"`
+	RSTStreams          uint64 `json:"rstStreams"`
+	GoAwayFrames        uint64 `json:"goAwayFrames"`
 	ActiveConnections   int64  `json:"activeConnections"`
 	ActiveRequests      int64  `json:"activeRequests"`
 }
@@ -97,6 +101,8 @@ func (s *Server) Stats() Stats {
 		RejectedRequests:    s.rejectedReqs.Load(),
 		ProtocolErrors:      s.protocolErrors.Load(),
 		GoAwaySent:          s.goAwaySent.Load(),
+		RSTStreams:          s.rstStreams.Load(),
+		GoAwayFrames:        s.goAwayFrames.Load(),
 		ActiveConnections:   s.activeConns.Load(),
 		ActiveRequests:      s.activeRequests.Load(),
 	}
@@ -356,7 +362,23 @@ func (s *Server) serveConn(ctx context.Context, protocolServer *xhttp2.Server, b
 	}
 	s.protocolConns[conn] = struct{}{}
 	s.mu.Unlock()
-	protocolServer.ServeConn(conn, &xhttp2.ServeConnOpts{
+	observed := &observedConn{
+		Conn:       conn,
+		readParser: frameParser{prefaceRemain: 24},
+		onFrame: func(direction frameDirection, frameType byte) {
+			if direction == frameRead && frameType == frameTypeRSTStream {
+				s.rstStreams.Add(1)
+			}
+			if direction == frameWrite && frameType == frameTypeGoAway {
+				s.goAwayFrames.Add(1)
+			}
+		},
+	}
+	var protocolConn net.Conn = observed
+	if tlsConn, ok := conn.(*tls.Conn); ok {
+		protocolConn = &observedTLSConn{observedConn: observed, state: tlsConn.ConnectionState()}
+	}
+	protocolServer.ServeConn(protocolConn, &xhttp2.ServeConnOpts{
 		Context: ctx, Handler: handler, BaseConfig: baseConfig,
 	})
 }

transport/native/HTTP2.md

@@ -22,7 +22,7 @@
 
 ## 必须可观测的状态
 
-HTTP/2 版本需要在 native `Stats` 之外至少记录：活动 stream、被拒绝 stream、连接级 protocol error、RST_STREAM、GOAWAY 发送次数，以及因流控或并发闸门触发的拒绝。实验 Runner 当前已通过 `ActiveRequests`、`RejectedRequests`、`ProtocolErrors` 和 `GoAwaySent` 暴露活动/拒绝/协议错误/优雅关闭快照；RST_STREAM 的独立计数仍属于进入 native 前的补齐项。计数必须是原子快照，不能在 metrics handler 中锁住连接状态机。
+HTTP/2 版本需要在 native `Stats` 之外至少记录：活动 stream、被拒绝 stream、连接级 protocol error、RST_STREAM、GOAWAY 发送次数，以及因流控或并发闸门触发的拒绝。实验 Runner 当前已通过 `ActiveRequests`、`RejectedRequests`、`ProtocolErrors`、`RSTStreams`、`GoAwaySent` 和 `GoAwayFrames` 暴露活动/拒绝/协议错误/客户端重置/优雅关闭快照；帧观察器只解析 9 字节头部并跳过 payload，不会缓存大 body。计数必须是原子快照，不能在 metrics handler 中锁住连接状态机。
 
 ## 进入实现前的验收矩阵