feat: expose native server stats

transport/native/server.go

@@ -62,17 +62,42 @@ type Server struct {
 	wg             sync.WaitGroup
 	connSlots      chan struct{}
 	requestSlots   chan struct{}
+	accepted       atomic.Uint64
+	activeConns    atomic.Int64
+	activeRequests atomic.Int64
 	readerPool     sync.Pool
 	writerPool     sync.Pool
 	connectionPool sync.Pool
 }
 
+// Stats is a point-in-time view of the native server lifecycle. Counters are
+// monotonic unless their name starts with Active.
+type Stats struct {
+	AcceptedConnections uint64 `json:"acceptedConnections"`
+	ActiveConnections   int64  `json:"activeConnections"`
+	ActiveRequests      int64  `json:"activeRequests"`
+}
+
+// Stats returns native connection and request counters without taking the
+// server lifecycle mutex, so it is safe to call from a metrics endpoint.
+func (s *Server) Stats() Stats {
+	if s == nil {
+		return Stats{}
+	}
+	return Stats{
+		AcceptedConnections: s.accepted.Load(),
+		ActiveConnections:   s.activeConns.Load(),
+		ActiveRequests:      s.activeRequests.Load(),
+	}
+}
+
 type connection struct {
 	server        *Server
 	conn          net.Conn
 	active        atomic.Bool
 	closeWhenIdle atomic.Bool
 	closed        atomic.Bool
+	counted       bool
 	slotHeld      bool
 	closeDone     chan struct{}
 }
@@ -222,6 +247,7 @@ func (s *Server) Serve(ctx context.Context, listener net.Listener) error {
 			serveErr = err
 			break
 		}
+		s.accepted.Add(1)
 		if slots := s.connSlots; slots != nil {
 			select {
 			case slots <- struct{}{}:
@@ -236,8 +262,10 @@ func (s *Server) Serve(ctx context.Context, listener net.Listener) error {
 		state.active.Store(false)
 		state.closeWhenIdle.Store(false)
 		state.closed.Store(false)
+		state.counted = true
 		state.slotHeld = s.connSlots != nil
 		state.closeDone = make(chan struct{})
+		s.activeConns.Add(1)
 		s.mu.Lock()
 		if s.closing {
 			s.mu.Unlock()
@@ -379,6 +407,7 @@ func (s *Server) serveConnection(baseCtx context.Context, state *connection) {
 		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)
 		writer.configure(request, state.conn, closeAfter, s.WriteTimeout, cancel)
@@ -392,6 +421,7 @@ func (s *Server) serveConnection(baseCtx context.Context, state *connection) {
 				writer.reset("")
 				s.writerPool.Put(writer)
 				state.active.Store(false)
+				s.activeRequests.Add(-1)
 				return
 			default:
 				writer.Header().Set("Content-Type", "text/plain; charset=utf-8")
@@ -403,6 +433,7 @@ func (s *Server) serveConnection(baseCtx context.Context, state *connection) {
 				writer.reset("")
 				s.writerPool.Put(writer)
 				state.active.Store(false)
+				s.activeRequests.Add(-1)
 				if closeAfter || closing {
 					return
 				}
@@ -439,6 +470,7 @@ func (s *Server) serveConnection(baseCtx context.Context, state *connection) {
 				writer.reset("")
 				s.writerPool.Put(writer)
 				state.active.Store(false)
+				s.activeRequests.Add(-1)
 				return
 			}
 			// Request bodies are streamed into the handler, so a malformed
@@ -460,6 +492,7 @@ func (s *Server) serveConnection(baseCtx context.Context, state *connection) {
 			writer.reset("")
 			s.writerPool.Put(writer)
 			state.active.Store(false)
+			s.activeRequests.Add(-1)
 			return
 		}
 		if requestSlot {
@@ -469,6 +502,7 @@ func (s *Server) serveConnection(baseCtx context.Context, state *connection) {
 		writer.reset("")
 		s.writerPool.Put(writer)
 		state.active.Store(false)
+		s.activeRequests.Add(-1)
 		if shouldClose {
 			return
 		}
@@ -482,6 +516,10 @@ func (c *connection) close() {
 		c.server.mu.Lock()
 		delete(c.server.conns, c)
 		c.server.mu.Unlock()
+		if c.counted {
+			c.server.activeConns.Add(-1)
+			c.counted = false
+		}
 	}
 }
 

transport/native/server_test.go

@@ -789,6 +789,57 @@ func TestServerConnectionLimitReleasesAfterRequest(t *testing.T) {
 	}
 }
 
+func TestServerStatsTrackLifecycle(t *testing.T) {
+	started := make(chan struct{})
+	release := make(chan struct{})
+	application := omni.NewWithRouter(netrouter.New())
+	application.GET("/hold", func(ctx *omni.Context) error {
+		close(started)
+		<-release
+		return ctx.OK("done")
+	})
+	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()
+	waitForStats := func(check func(native.Stats) bool) {
+		t.Helper()
+		deadline := time.Now().Add(time.Second)
+		for time.Now().Before(deadline) {
+			if check(server.Stats()) {
+				return
+			}
+			time.Sleep(time.Millisecond)
+		}
+		t.Fatalf("stats condition not reached: %+v", server.Stats())
+	}
+	waitForStats(func(stats native.Stats) bool { return stats.ActiveConnections == 1 && stats.AcceptedConnections >= 1 })
+	if _, err := io.WriteString(conn, "GET /hold HTTP/1.1\r\nHost: test\r\nConnection: close\r\n\r\n"); err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case <-started:
+	case <-time.After(time.Second):
+		t.Fatal("hold handler did not start")
+	}
+	waitForStats(func(stats native.Stats) bool { return stats.ActiveRequests == 1 })
+	close(release)
+	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.StatusOK {
+		t.Fatalf("unexpected hold response: %d", response.StatusCode)
+	}
+	waitForStats(func(stats native.Stats) bool { return stats.ActiveRequests == 0 })
+	waitForStats(func(stats native.Stats) bool { return stats.ActiveConnections == 0 })
+}
+
 func TestServerConnectionChurnReusesStateSafely(t *testing.T) {
 	application := omni.NewWithRouter(netrouter.New())
 	application.GET("/health", func(ctx *omni.Context) error { return ctx.OK("ok") })