fix(server): prevent control replacement lifecycle leaks (#5424)

.github/workflows/build-and-push-image.yml

@@ -65,7 +65,7 @@ jobs:
         with:
           context: .
           file: ./dockerfiles/Dockerfile-for-frpc
-          platforms: linux/amd64,linux/arm/v7,linux/arm64,linux/ppc64le,linux/s390x
+          platforms: linux/amd64,linux/arm/v7,linux/arm64,linux/ppc64le
           push: true
           tags: |
             ${{ env.TAG_FRPC }}
@@ -76,7 +76,7 @@ jobs:
         with:
           context: .
           file: ./dockerfiles/Dockerfile-for-frps
-          platforms: linux/amd64,linux/arm/v7,linux/arm64,linux/ppc64le,linux/s390x
+          platforms: linux/amd64,linux/arm/v7,linux/arm64,linux/ppc64le
           push: true
           tags: |
             ${{ env.TAG_FRPS }}

Makefile.cross-compiles

@@ -9,7 +9,7 @@ all: build
 build: app
 
 app:
-	@$(foreach n, $(os-archs), \
+	@set -e; $(foreach n, $(os-archs), \
 		os=$(shell echo "$(n)" | cut -d : -f 1); \
 		arch=$(shell echo "$(n)" | cut -d : -f 2); \
 		extra=$(shell echo "$(n)" | cut -d : -f 3); \

README.md

@@ -2,7 +2,6 @@
 
 [![Build Status](https://circleci.com/gh/fatedier/frp.svg?style=shield)](https://circleci.com/gh/fatedier/frp)
 [![GitHub release](https://img.shields.io/github/tag/fatedier/frp.svg?label=release)](https://github.com/fatedier/frp/releases)
-[![Go Report Card](https://goreportcard.com/badge/github.com/fatedier/frp)](https://goreportcard.com/report/github.com/fatedier/frp)
 [![GitHub Releases Stats](https://img.shields.io/github/downloads/fatedier/frp/total.svg?logo=github)](https://somsubhra.github.io/github-release-stats/?username=fatedier&repository=frp)
 
 [README](README.md) | [中文文档](README_zh.md)

Release.md

@@ -1,10 +1,3 @@
-## Features
-
-* Added dashboard API v2 pagination endpoints for users, clients, and proxies.
-* The frps dashboard Clients and Proxies pages now use API v2 pagination and server-side search, including proxy type filtering and searchable proxy spec fields such as remote ports, custom domains, and subdomains.
-
 ## Fixes
 
-* WebSocket and WSS tunnel payloads are now sent as binary frames, avoiding disconnects through RFC-compliant intermediaries that validate text frames as UTF-8.
-* The `tls2raw` client plugin now writes the proxy protocol header to the local raw connection when proxy protocol is enabled.
-* frpc now rejects duplicate proxy and visitor names in config files instead of silently overwriting earlier entries.
+* Fixed control-session replacement leaks when frpc reconnects through a half-open TCP multiplexed connection.

go.mod

@@ -5,7 +5,7 @@ go 1.25.0
 require (
 	github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5
 	github.com/coreos/go-oidc/v3 v3.14.1
-	github.com/fatedier/golib v0.7.0
+	github.com/fatedier/golib v0.7.2-0.20260715094513-2780ae1d8cdb
 	github.com/google/uuid v1.6.0
 	github.com/gorilla/mux v1.8.1
 	github.com/gorilla/websocket v1.5.0
@@ -13,7 +13,6 @@ require (
 	github.com/onsi/ginkgo/v2 v2.23.4
 	github.com/onsi/gomega v1.36.3
 	github.com/pelletier/go-toml/v2 v2.2.0
-	github.com/pion/stun/v3 v3.1.1
 	github.com/pires/go-proxyproto v0.7.0
 	github.com/prometheus/client_golang v1.19.1
 	github.com/quic-go/quic-go v0.55.0
@@ -53,9 +52,6 @@ require (
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/klauspost/cpuid/v2 v2.2.6 // indirect
 	github.com/klauspost/reedsolomon v1.12.0 // indirect
-	github.com/pion/dtls/v3 v3.0.10 // indirect
-	github.com/pion/logging v0.2.4 // indirect
-	github.com/pion/transport/v4 v4.0.1 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/prometheus/client_model v0.5.0 // indirect
@@ -67,7 +63,6 @@ require (
 	github.com/tidwall/pretty v1.2.0 // indirect
 	github.com/tjfoc/gmsm v1.4.1 // indirect
 	github.com/vishvananda/netns v0.0.4 // indirect
-	github.com/wlynxg/anet v0.0.5 // indirect
 	go.uber.org/automaxprocs v1.6.0 // indirect
 	golang.org/x/mod v0.33.0 // indirect
 	golang.org/x/text v0.35.0 // indirect

go.sum

@@ -20,8 +20,8 @@ github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSs
 github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
 github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1mIlRU8Am5FuJP05cCM98=
 github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
-github.com/fatedier/golib v0.7.0 h1:tMDF9ObcwVt59VUHroJOzHQjVFPLymZVMpGm9WAVwhY=
-github.com/fatedier/golib v0.7.0/go.mod h1:ArUGvPg2cOw/py2RAuBt46nNZH2VQ5Z70p109MAZpJw=
+github.com/fatedier/golib v0.7.2-0.20260715094513-2780ae1d8cdb h1:6gcaE769mFb1G/cNV6lgzmwIYkNXmhj4pmcddllfmB4=
+github.com/fatedier/golib v0.7.2-0.20260715094513-2780ae1d8cdb/go.mod h1:ArUGvPg2cOw/py2RAuBt46nNZH2VQ5Z70p109MAZpJw=
 github.com/fatedier/yamux v0.0.0-20250825093530-d0154be01cd6 h1:u92UUy6FURPmNsMBUuongRWC0rBqN6gd01Dzu+D21NE=
 github.com/fatedier/yamux v0.0.0-20250825093530-d0154be01cd6/go.mod h1:c5/tk6G0dSpXGzJN7Wk1OEie8grdSJAmeawId9Zvd34=
 github.com/go-jose/go-jose/v4 v4.0.5 h1:M6T8+mKZl/+fNNuFHvGIzDz7BTLQPIounk/b9dw3AaE=
@@ -78,14 +78,6 @@ github.com/onsi/gomega v1.36.3 h1:hID7cr8t3Wp26+cYnfcjR6HpJ00fdogN6dqZ1t6IylU=
 github.com/onsi/gomega v1.36.3/go.mod h1:8D9+Txp43QWKhM24yyOBEdpkzN8FvJyAwecBgsU4KU0=
 github.com/pelletier/go-toml/v2 v2.2.0 h1:QLgLl2yMN7N+ruc31VynXs1vhMZa7CeHHejIeBAsoHo=
 github.com/pelletier/go-toml/v2 v2.2.0/go.mod h1:1t835xjRzz80PqgE6HHgN2JOsmgYu/h4qDAS4n929Rs=
-github.com/pion/dtls/v3 v3.0.10 h1:k9ekkq1kaZoxnNEbyLKI8DI37j/Nbk1HWmMuywpQJgg=
-github.com/pion/dtls/v3 v3.0.10/go.mod h1:YEmmBYIoBsY3jmG56dsziTv/Lca9y4Om83370CXfqJ8=
-github.com/pion/logging v0.2.4 h1:tTew+7cmQ+Mc1pTBLKH2puKsOvhm32dROumOZ655zB8=
-github.com/pion/logging v0.2.4/go.mod h1:DffhXTKYdNZU+KtJ5pyQDjvOAh/GsNSyv1lbkFbe3so=
-github.com/pion/stun/v3 v3.1.1 h1:CkQxveJ4xGQjulGSROXbXq94TAWu8gIX2dT+ePhUkqw=
-github.com/pion/stun/v3 v3.1.1/go.mod h1:qC1DfmcCTQjl9PBaMa5wSn3x9IPmKxSdcCsxBcDBndM=
-github.com/pion/transport/v4 v4.0.1 h1:sdROELU6BZ63Ab7FrOLn13M6YdJLY20wldXW2Cu2k8o=
-github.com/pion/transport/v4 v4.0.1/go.mod h1:nEuEA4AD5lPdcIegQDpVLgNoDGreqM/YqmEx3ovP4jM=
 github.com/pires/go-proxyproto v0.7.0 h1:IukmRewDQFWC7kfnb66CSomk2q/seBuilHBYFwyq0Hs=
 github.com/pires/go-proxyproto v0.7.0/go.mod h1:Vz/1JPY/OACxWGQNIRY2BeyDmpoaWmEP40O9LbuiFR4=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
@@ -146,8 +138,6 @@ github.com/vishvananda/netlink v1.3.0 h1:X7l42GfcV4S6E4vHTsw48qbrV+9PVojNfIhZcwQ
 github.com/vishvananda/netlink v1.3.0/go.mod h1:i6NetklAujEcC6fK0JPjT8qSwWyO0HLn4UKG+hGqeJs=
 github.com/vishvananda/netns v0.0.4 h1:Oeaw1EM2JMxD51g9uhtC0D7erkIjgmj8+JZc26m1YX8=
 github.com/vishvananda/netns v0.0.4/go.mod h1:SpkAiCQRtJ6TvvxPnOSyH3BMl6unz3xZlaprSwhNNJM=
-github.com/wlynxg/anet v0.0.5 h1:J3VJGi1gvo0JwZ/P1/Yc/8p63SoW98B5dHkYDmpgvvU=
-github.com/wlynxg/anet v0.0.5/go.mod h1:eay5PRQr7fIVAMbTbchTnO9gG65Hg/uYGdc7mguHxoA=
 github.com/xtaci/kcp-go/v5 v5.6.13 h1:FEjtz9+D4p8t2x4WjciGt/jsIuhlWjjgPCCWjrVR4Hk=
 github.com/xtaci/kcp-go/v5 v5.6.13/go.mod h1:75S1AKYYzNUSXIv30h+jPKJYZUwqpfvLshu63nCNSOM=
 github.com/xtaci/lossyconn v0.0.0-20200209145036-adba10fffc37 h1:EWU6Pktpas0n8lLQwDsRyZfmkPeRbdgPtW609es+/9E=

pkg/metrics/mem/server.go

@@ -239,9 +239,11 @@ func toProxyStats(name string, proxyStats *ProxyStatistics) *ProxyStats {
 	}
 	if !proxyStats.LastStartTime.IsZero() {
 		ps.LastStartTime = proxyStats.LastStartTime.Format("01-02 15:04:05")
+		ps.LastStartAt = proxyStats.LastStartTime.Unix()
 	}
 	if !proxyStats.LastCloseTime.IsZero() {
 		ps.LastCloseTime = proxyStats.LastCloseTime.Format("01-02 15:04:05")
+		ps.LastCloseAt = proxyStats.LastCloseTime.Unix()
 	}
 	return ps
 }

pkg/metrics/mem/server_test.go

@@ -22,6 +22,12 @@ func TestServerMetricsUsesClockForProxyTimestamps(t *testing.T) {
 	clk.SetTime(closedAt)
 	metrics.CloseProxy("proxy", "tcp")
 	require.Equal(closedAt, metrics.info.ProxyStatistics["proxy"].LastCloseTime)
+
+	stats := metrics.GetProxyByName("proxy")
+	require.Equal(start.Format("01-02 15:04:05"), stats.LastStartTime)
+	require.Equal(closedAt.Format("01-02 15:04:05"), stats.LastCloseTime)
+	require.Equal(start.Unix(), stats.LastStartAt)
+	require.Equal(closedAt.Unix(), stats.LastCloseAt)
 }
 
 func TestServerMetricsClearUselessInfoUsesClock(t *testing.T) {

pkg/metrics/mem/types.go

@@ -41,6 +41,8 @@ type ProxyStats struct {
 	TodayTrafficOut int64
 	LastStartTime   string
 	LastCloseTime   string
+	LastStartAt     int64
+	LastCloseAt     int64
 	CurConns        int64
 }
 

pkg/nathole/discovery.go

@@ -15,31 +15,24 @@
 package nathole
 
 import (
+	"errors"
 	"fmt"
 	"net"
 	"time"
 
-	"github.com/pion/stun/v3"
+	"github.com/fatedier/golib/net/stun"
 )
 
 var responseTimeout = 3 * time.Second
 
-type Message struct {
-	Body []byte
-	Addr string
-}
-
 // If the localAddr is empty, it will listen on a random port.
 func Discover(stunServers []string, localAddr string) ([]string, net.Addr, error) {
-	// create a discoverConn and get response from messageChan
 	discoverConn, err := listen(localAddr)
 	if err != nil {
 		return nil, nil, err
 	}
 	defer discoverConn.Close()
 
-	go discoverConn.readLoop()
-
 	addresses := make([]string, 0, len(stunServers))
 	for _, addr := range stunServers {
 		// get external address from stun server
@@ -58,10 +51,9 @@ type stunResponse struct {
 }
 
 type discoverConn struct {
-	conn *net.UDPConn
-
-	localAddr   net.Addr
-	messageChan chan *Message
+	conn      *net.UDPConn
+	client    *stun.Client
+	localAddr net.Addr
 }
 
 func listen(localAddr string) (*discoverConn, error) {
@@ -77,82 +69,50 @@ func listen(localAddr string) (*discoverConn, error) {
 	if err != nil {
 		return nil, err
 	}
+	client, err := stun.NewClient(conn)
+	if err != nil {
+		_ = conn.Close()
+		return nil, err
+	}
 
 	return &discoverConn{
-		conn:        conn,
-		localAddr:   conn.LocalAddr(),
-		messageChan: make(chan *Message, 10),
+		conn:      conn,
+		client:    client,
+		localAddr: conn.LocalAddr(),
 	}, nil
 }
 
 func (c *discoverConn) Close() error {
-	if c.messageChan != nil {
-		close(c.messageChan)
-		c.messageChan = nil
-	}
 	return c.conn.Close()
 }
 
-func (c *discoverConn) readLoop() {
-	for {
-		buf := make([]byte, 1024)
-		n, addr, err := c.conn.ReadFromUDP(buf)
-		if err != nil {
-			return
-		}
-		buf = buf[:n]
-
-		c.messageChan <- &Message{
-			Body: buf,
-			Addr: addr.String(),
-		}
-	}
-}
-
 func (c *discoverConn) doSTUNRequest(addr string) (*stunResponse, error) {
 	serverAddr, err := net.ResolveUDPAddr("udp4", addr)
 	if err != nil {
 		return nil, err
 	}
-	request, err := stun.Build(stun.TransactionID, stun.BindingRequest)
+	transaction, err := stun.NewBindingTransaction(serverAddr)
 	if err != nil {
 		return nil, err
 	}
-
-	if err = request.NewTransactionID(); err != nil {
+	if err := c.conn.SetReadDeadline(time.Now().Add(responseTimeout)); err != nil {
 		return nil, err
 	}
-	if _, err := c.conn.WriteTo(request.Raw, serverAddr); err != nil {
-		return nil, err
-	}
-
-	var m stun.Message
-	select {
-	case msg := <-c.messageChan:
-		m.Raw = msg.Body
-		if err := m.Decode(); err != nil {
-			return nil, err
+	response, err := c.client.Do(transaction)
+	if err != nil {
+		var netErr net.Error
+		if errors.As(err, &netErr) && netErr.Timeout() {
+			return nil, fmt.Errorf("wait response from stun server timeout")
 		}
-	case <-time.After(responseTimeout):
-		return nil, fmt.Errorf("wait response from stun server timeout")
+		return nil, err
 	}
-	xorAddrGetter := &stun.XORMappedAddress{}
-	mappedAddrGetter := &stun.MappedAddress{}
-	changedAddrGetter := ChangedAddress{}
-	otherAddrGetter := &stun.OtherAddress{}
 
 	resp := &stunResponse{}
-	if err := mappedAddrGetter.GetFrom(&m); err == nil {
-		resp.externalAddr = mappedAddrGetter.String()
+	if response.MappedAddr != nil {
+		resp.externalAddr = response.MappedAddr.String()
 	}
-	if err := xorAddrGetter.GetFrom(&m); err == nil {
-		resp.externalAddr = xorAddrGetter.String()
-	}
-	if err := changedAddrGetter.GetFrom(&m); err == nil {
-		resp.otherAddr = changedAddrGetter.String()
-	}
-	if err := otherAddrGetter.GetFrom(&m); err == nil {
-		resp.otherAddr = otherAddrGetter.String()
+	if response.OtherAddr != nil {
+		resp.otherAddr = response.OtherAddr.String()
 	}
 	return resp, nil
 }

pkg/nathole/discovery_test.go

@@ -0,0 +1,382 @@
+package nathole
+
+import (
+	"encoding/binary"
+	"errors"
+	"fmt"
+	"net"
+	"testing"
+	"time"
+
+	"github.com/fatedier/golib/net/stun"
+	"github.com/stretchr/testify/require"
+)
+
+const (
+	testBindingRequest  = 0x0001
+	testBindingSuccess  = 0x0101
+	testBindingError    = 0x0111
+	testMagicCookie     = 0x2112a442
+	testAttrMapped      = 0x0001
+	testAttrChanged     = 0x0005
+	testAttrErrorCode   = 0x0009
+	testAttrXORMapped   = 0x0020
+	testAttrOther       = 0x802c
+	testSTUNHeaderSize  = 20
+	testSTUNServerLimit = time.Second
+)
+
+type testSTUNAttribute struct {
+	typ   uint16
+	value []byte
+}
+
+type testSTUNExchange struct {
+	source *net.UDPAddr
+	err    error
+}
+
+func listenTestUDP4(t *testing.T) *net.UDPConn {
+	t.Helper()
+
+	conn, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)})
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = conn.Close() })
+	return conn
+}
+
+func serveOneSTUNRequest(
+	server *net.UDPConn,
+	buildResponse func([]byte, *net.UDPAddr) ([]byte, error),
+) <-chan testSTUNExchange {
+	done := make(chan testSTUNExchange, 1)
+	go func() {
+		if err := server.SetDeadline(time.Now().Add(testSTUNServerLimit)); err != nil {
+			done <- testSTUNExchange{err: err}
+			return
+		}
+		buffer := make([]byte, 1024)
+		n, source, err := server.ReadFromUDP(buffer)
+		if err == nil && buildResponse != nil {
+			var response []byte
+			response, err = buildResponse(buffer[:n], source)
+			if err == nil && response != nil {
+				_, err = server.WriteToUDP(response, source)
+			}
+		}
+		done <- testSTUNExchange{source: source, err: err}
+	}()
+	return done
+}
+
+func waitSTUNExchange(t *testing.T, done <-chan testSTUNExchange) *net.UDPAddr {
+	t.Helper()
+
+	select {
+	case exchange := <-done:
+		require.NoError(t, exchange.err)
+		return exchange.source
+	case <-time.After(testSTUNServerLimit):
+		t.Fatal("timed out waiting for local STUN server")
+		return nil
+	}
+}
+
+func makeTestSTUNResponse(request []byte, typ uint16, attributes ...testSTUNAttribute) ([]byte, error) {
+	if len(request) != testSTUNHeaderSize || binary.BigEndian.Uint16(request[0:2]) != testBindingRequest ||
+		binary.BigEndian.Uint32(request[4:8]) != testMagicCookie {
+		return nil, fmt.Errorf("invalid Binding request")
+	}
+
+	length := 0
+	for _, attribute := range attributes {
+		length += 4 + (len(attribute.value)+3)&^3
+	}
+	response := make([]byte, testSTUNHeaderSize, testSTUNHeaderSize+length)
+	binary.BigEndian.PutUint16(response[0:2], typ)
+	binary.BigEndian.PutUint16(response[2:4], uint16(length))
+	binary.BigEndian.PutUint32(response[4:8], testMagicCookie)
+	copy(response[8:20], request[8:20])
+
+	for _, attribute := range attributes {
+		start := len(response)
+		paddedLength := (len(attribute.value) + 3) &^ 3
+		response = append(response, make([]byte, 4+paddedLength)...)
+		binary.BigEndian.PutUint16(response[start:start+2], attribute.typ)
+		binary.BigEndian.PutUint16(response[start+2:start+4], uint16(len(attribute.value)))
+		copy(response[start+4:], attribute.value)
+	}
+	return response, nil
+}
+
+func testIPv4AddressValue(ip net.IP, port int, xor bool) []byte {
+	value := make([]byte, 8)
+	value[1] = 0x01
+	binary.BigEndian.PutUint16(value[2:4], uint16(port))
+	copy(value[4:], ip.To4())
+	if xor {
+		binary.BigEndian.PutUint16(value[2:4], binary.BigEndian.Uint16(value[2:4])^uint16(testMagicCookie>>16))
+		for i := range 4 {
+			value[4+i] ^= byte(uint32(testMagicCookie) >> uint(24-8*i))
+		}
+	}
+	return value
+}
+
+func TestDiscoverReusesLocalPortAndPreservesNATClassification(t *testing.T) {
+	tests := []struct {
+		name             string
+		secondMapped     string
+		secondMappedPort int
+		wantNATType      string
+		wantBehavior     string
+	}{
+		{
+			name:             "same mapped address",
+			secondMapped:     "198.51.100.10:40000",
+			secondMappedPort: 40000,
+			wantNATType:      EasyNAT,
+			wantBehavior:     BehaviorNoChange,
+		},
+		{
+			name:             "different mapped port",
+			secondMapped:     "198.51.100.10:40001",
+			secondMappedPort: 40001,
+			wantNATType:      HardNAT,
+			wantBehavior:     BehaviorPortChanged,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			primary := listenTestUDP4(t)
+			alternate := listenTestUDP4(t)
+			alternateAddr := alternate.LocalAddr().(*net.UDPAddr)
+
+			primaryDone := serveOneSTUNRequest(primary, func(request []byte, _ *net.UDPAddr) ([]byte, error) {
+				return makeTestSTUNResponse(request, testBindingSuccess,
+					testSTUNAttribute{typ: testAttrXORMapped, value: testIPv4AddressValue(net.ParseIP("198.51.100.10"), 40000, true)},
+					testSTUNAttribute{typ: testAttrOther, value: testIPv4AddressValue(alternateAddr.IP, alternateAddr.Port, false)},
+				)
+			})
+			alternateDone := serveOneSTUNRequest(alternate, func(request []byte, _ *net.UDPAddr) ([]byte, error) {
+				return makeTestSTUNResponse(request, testBindingSuccess,
+					testSTUNAttribute{typ: testAttrXORMapped, value: testIPv4AddressValue(net.ParseIP("198.51.100.10"), tt.secondMappedPort, true)},
+				)
+			})
+
+			addresses, localAddr, err := Discover([]string{primary.LocalAddr().String()}, "")
+			require.NoError(t, err)
+			require.Equal(t, []string{"198.51.100.10:40000", tt.secondMapped}, addresses)
+
+			primarySource := waitSTUNExchange(t, primaryDone)
+			alternateSource := waitSTUNExchange(t, alternateDone)
+			require.Equal(t, primarySource.Port, alternateSource.Port)
+			require.Equal(t, localAddr.(*net.UDPAddr).Port, primarySource.Port)
+
+			feature, err := ClassifyNATFeature(addresses, nil)
+			require.NoError(t, err)
+			require.Equal(t, tt.wantNATType, feature.NatType)
+			require.Equal(t, tt.wantBehavior, feature.Behavior)
+		})
+	}
+}
+
+func TestDoSTUNRequestMapsLegacyAndModernAddresses(t *testing.T) {
+	tests := []struct {
+		name         string
+		attributes   []testSTUNAttribute
+		wantExternal string
+		wantOther    string
+	}{
+		{
+			name: "legacy",
+			attributes: []testSTUNAttribute{
+				{typ: testAttrMapped, value: testIPv4AddressValue(net.ParseIP("192.0.2.1"), 1000, false)},
+				{typ: testAttrChanged, value: testIPv4AddressValue(net.ParseIP("192.0.2.2"), 2000, false)},
+			},
+			wantExternal: "192.0.2.1:1000",
+			wantOther:    "192.0.2.2:2000",
+		},
+		{
+			name: "modern takes precedence",
+			attributes: []testSTUNAttribute{
+				{typ: testAttrMapped, value: testIPv4AddressValue(net.ParseIP("192.0.2.1"), 1000, false)},
+				{typ: testAttrXORMapped, value: testIPv4AddressValue(net.ParseIP("198.51.100.1"), 3000, true)},
+				{typ: testAttrChanged, value: testIPv4AddressValue(net.ParseIP("192.0.2.2"), 2000, false)},
+				{typ: testAttrOther, value: testIPv4AddressValue(net.ParseIP("198.51.100.2"), 4000, false)},
+			},
+			wantExternal: "198.51.100.1:3000",
+			wantOther:    "198.51.100.2:4000",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			server := listenTestUDP4(t)
+			done := serveOneSTUNRequest(server, func(request []byte, _ *net.UDPAddr) ([]byte, error) {
+				return makeTestSTUNResponse(request, testBindingSuccess, tt.attributes...)
+			})
+			conn, err := listen("")
+			require.NoError(t, err)
+			t.Cleanup(func() { _ = conn.Close() })
+
+			response, err := conn.doSTUNRequest(server.LocalAddr().String())
+			require.NoError(t, err)
+			require.Equal(t, tt.wantExternal, response.externalAddr)
+			require.Equal(t, tt.wantOther, response.otherAddr)
+			waitSTUNExchange(t, done)
+		})
+	}
+}
+
+func TestSTUNResponseErrorsAndMissingAddresses(t *testing.T) {
+	tests := []struct {
+		name          string
+		buildResponse func([]byte, *net.UDPAddr) ([]byte, error)
+		request       func(*discoverConn, string) error
+		checkError    func(*testing.T, error)
+	}{
+		{
+			name: "correlated malformed response",
+			buildResponse: func(request []byte, _ *net.UDPAddr) ([]byte, error) {
+				response, err := makeTestSTUNResponse(request, testBindingSuccess)
+				if err == nil {
+					binary.BigEndian.PutUint16(response[2:4], 4)
+				}
+				return response, err
+			},
+			request: func(conn *discoverConn, server string) error {
+				_, err := conn.doSTUNRequest(server)
+				return err
+			},
+			checkError: func(t *testing.T, err error) {
+				require.ErrorIs(t, err, stun.ErrMalformedResponse)
+			},
+		},
+		{
+			name: "Binding error response",
+			buildResponse: func(request []byte, _ *net.UDPAddr) ([]byte, error) {
+				return makeTestSTUNResponse(request, testBindingError, testSTUNAttribute{
+					typ:   testAttrErrorCode,
+					value: []byte{0, 0, 4, 20, 'U', 'n', 'k', 'n', 'o', 'w', 'n'},
+				})
+			},
+			request: func(conn *discoverConn, server string) error {
+				_, err := conn.doSTUNRequest(server)
+				return err
+			},
+			checkError: func(t *testing.T, err error) {
+				var responseErr *stun.ResponseError
+				require.ErrorAs(t, err, &responseErr)
+				require.Equal(t, 420, responseErr.Code)
+			},
+		},
+		{
+			name: "missing mapped address",
+			buildResponse: func(request []byte, _ *net.UDPAddr) ([]byte, error) {
+				return makeTestSTUNResponse(request, testBindingSuccess,
+					testSTUNAttribute{typ: testAttrOther, value: testIPv4AddressValue(net.ParseIP("192.0.2.2"), 2000, false)},
+				)
+			},
+			request: func(conn *discoverConn, server string) error {
+				_, err := conn.discoverFromStunServer(server)
+				return err
+			},
+			checkError: func(t *testing.T, err error) {
+				require.EqualError(t, err, "no external address found")
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			server := listenTestUDP4(t)
+			done := serveOneSTUNRequest(server, tt.buildResponse)
+			conn, err := listen("")
+			require.NoError(t, err)
+			t.Cleanup(func() { _ = conn.Close() })
+
+			err = tt.request(conn, server.LocalAddr().String())
+			tt.checkError(t, err)
+			waitSTUNExchange(t, done)
+		})
+	}
+
+	t.Run("missing other address", func(t *testing.T) {
+		server := listenTestUDP4(t)
+		done := serveOneSTUNRequest(server, func(request []byte, _ *net.UDPAddr) ([]byte, error) {
+			return makeTestSTUNResponse(request, testBindingSuccess,
+				testSTUNAttribute{typ: testAttrXORMapped, value: testIPv4AddressValue(net.ParseIP("198.51.100.1"), 3000, true)},
+			)
+		})
+
+		_, err := Prepare([]string{server.LocalAddr().String()}, PrepareOptions{})
+		require.EqualError(t, err, "discover error: not enough addresses")
+		waitSTUNExchange(t, done)
+	})
+}
+
+func TestSTUNTimeoutUsesCallerDeadlineWithoutRetry(t *testing.T) {
+	originalTimeout := responseTimeout
+	responseTimeout = 50 * time.Millisecond
+	t.Cleanup(func() { responseTimeout = originalTimeout })
+
+	server := listenTestUDP4(t)
+	done := serveOneSTUNRequest(server, nil)
+	conn, err := listen("")
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = conn.Close() })
+
+	_, err = conn.doSTUNRequest(server.LocalAddr().String())
+	require.EqualError(t, err, "wait response from stun server timeout")
+	waitSTUNExchange(t, done)
+
+	require.NoError(t, server.SetReadDeadline(time.Now().Add(50*time.Millisecond)))
+	_, _, err = server.ReadFromUDP(make([]byte, 1))
+	var netErr net.Error
+	require.ErrorAs(t, err, &netErr)
+	require.True(t, netErr.Timeout())
+}
+
+func TestSTUNClientLeavesSocketAndDeadlineWithCaller(t *testing.T) {
+	originalTimeout := responseTimeout
+	responseTimeout = 100 * time.Millisecond
+	t.Cleanup(func() { responseTimeout = originalTimeout })
+
+	server := listenTestUDP4(t)
+	unrelated := listenTestUDP4(t)
+	done := serveOneSTUNRequest(server, func(request []byte, source *net.UDPAddr) ([]byte, error) {
+		response, err := makeTestSTUNResponse(request, testBindingSuccess,
+			testSTUNAttribute{typ: testAttrXORMapped, value: testIPv4AddressValue(net.ParseIP("198.51.100.5"), 5000, true)},
+		)
+		if err != nil {
+			return nil, err
+		}
+		if _, err := unrelated.WriteToUDP(response, source); err != nil {
+			return nil, err
+		}
+		return response, nil
+	})
+	conn, err := listen("")
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = conn.Close() })
+
+	response, err := conn.doSTUNRequest(server.LocalAddr().String())
+	require.NoError(t, err)
+	require.Equal(t, "198.51.100.5:5000", response.externalAddr)
+	waitSTUNExchange(t, done)
+
+	_, _, err = conn.conn.ReadFromUDP(make([]byte, 1))
+	var netErr net.Error
+	require.True(t, errors.As(err, &netErr))
+	require.True(t, netErr.Timeout())
+
+	require.NoError(t, conn.conn.SetDeadline(time.Time{}))
+	require.NoError(t, server.SetReadDeadline(time.Now().Add(testSTUNServerLimit)))
+	_, err = conn.conn.WriteToUDP([]byte{1}, server.LocalAddr().(*net.UDPAddr))
+	require.NoError(t, err)
+	_, source, err := server.ReadFromUDP(make([]byte, 1))
+	require.NoError(t, err)
+	require.Equal(t, conn.localAddr.(*net.UDPAddr).Port, source.Port)
+}

pkg/nathole/utils.go

@@ -18,10 +18,8 @@ import (
 	"bytes"
 	"fmt"
 	"net"
-	"strconv"
 
 	"github.com/fatedier/golib/crypto"
-	"github.com/pion/stun/v3"
 
 	"github.com/fatedier/frp/pkg/msg"
 )
@@ -48,20 +46,6 @@ func DecodeMessageInto(data, key []byte, m msg.Message) error {
 	return msg.ReadMsgInto(bytes.NewReader(buf), m)
 }
 
-type ChangedAddress struct {
-	IP   net.IP
-	Port int
-}
-
-func (s *ChangedAddress) GetFrom(m *stun.Message) error {
-	a := (*stun.MappedAddress)(s)
-	return a.GetFromAs(m, stun.AttrChangedAddress)
-}
-
-func (s *ChangedAddress) String() string {
-	return net.JoinHostPort(s.IP.String(), strconv.Itoa(s.Port))
-}
-
 func ListAllLocalIPs() ([]net.IP, error) {
 	addrs, err := net.InterfaceAddrs()
 	if err != nil {

pkg/util/version/version.go

@@ -14,7 +14,7 @@
 
 package version
 
-var version = "0.69.1"
+var version = "0.70.0"
 
 func Full() string {
 	return version

server/api_router.go

@@ -56,7 +56,8 @@ func (svr *Service) registerRouteHandlers(helper *httppkg.RouterRegisterHelper)
 	v2EncodedPathRouter.UseEncodedPath()
 	v2EncodedPathRouter.HandleFunc("/api/v2/clients/{key}", httppkg.MakeHTTPHandlerFuncV2(apiController.APIV2ClientDetail)).Methods("GET")
 	subRouter.HandleFunc("/api/v2/proxies", httppkg.MakeHTTPHandlerFuncV2(apiController.APIV2ProxyList)).Methods("GET")
-	subRouter.HandleFunc("/api/v2/proxies/{name}", httppkg.MakeHTTPHandlerFuncV2(apiController.APIV2ProxyDetail)).Methods("GET")
+	v2EncodedPathRouter.HandleFunc("/api/v2/proxies/{name}/traffic", httppkg.MakeHTTPHandlerFuncV2(apiController.APIV2ProxyTraffic)).Methods("GET")
+	v2EncodedPathRouter.HandleFunc("/api/v2/proxies/{name}", httppkg.MakeHTTPHandlerFuncV2(apiController.APIV2ProxyDetail)).Methods("GET")
 
 	// view
 	subRouter.Handle("/favicon.ico", http.FileServer(helper.AssetsFS)).Methods("GET")

server/control.go

@@ -17,6 +17,7 @@ package server
 import (
 	"context"
 	"fmt"
+	"net"
 	"runtime/debug"
 	"sync"
 	"sync/atomic"
@@ -40,55 +41,311 @@ import (
 	"github.com/fatedier/frp/server/registry"
 )
 
+type ControlID uint64
+
+var nextControlID atomic.Uint64
+
+type controlEntry struct {
+	ctl *Control
+	id  ControlID
+	// runMu serializes lifecycle and routing decisions for one run ID.
+	// Replacements inherit it; removing the entry releases the manager's reference.
+	runMu *sync.Mutex
+
+	registryOnline    bool
+	registryControlID ControlID
+}
+
 type ControlManager struct {
 	// controls indexed by run id
-	ctlsByRunID map[string]*Control
+	ctlsByRunID map[string]*controlEntry
+	registry    *registry.ClientRegistry
+	closed      bool
 
 	mu sync.RWMutex
 }
 
-func NewControlManager() *ControlManager {
+func NewControlManager(clientRegistry *registry.ClientRegistry) *ControlManager {
 	return &ControlManager{
-		ctlsByRunID: make(map[string]*Control),
+		ctlsByRunID: make(map[string]*controlEntry),
+		registry:    clientRegistry,
 	}
 }
 
-func (cm *ControlManager) Add(runID string, ctl *Control) (old *Control) {
-	cm.mu.Lock()
-	defer cm.mu.Unlock()
-
-	var ok bool
-	old, ok = cm.ctlsByRunID[runID]
-	if ok {
-		old.Replaced(ctl)
+// lockCurrentRun returns the current entry with its run gate held. It never
+// waits for the gate while holding cm.mu and revalidates the gate after waiting.
+// The global order is runMu, cm.mu, ctl.lifecycleMu, then registry locks.
+func (cm *ControlManager) lockCurrentRun(runID string, allowClosed bool) (*controlEntry, bool) {
+	cm.mu.RLock()
+	entry, ok := cm.ctlsByRunID[runID]
+	if cm.closed && !allowClosed {
+		ok = false
 	}
-	cm.ctlsByRunID[runID] = ctl
-	return
+	cm.mu.RUnlock()
+	if !ok {
+		return nil, false
+	}
+
+	runMu := entry.runMu
+	runMu.Lock()
+	cm.mu.RLock()
+	entry, ok = cm.ctlsByRunID[runID]
+	if (cm.closed && !allowClosed) || !ok || entry.runMu != runMu {
+		ok = false
+	}
+	cm.mu.RUnlock()
+	if !ok {
+		runMu.Unlock()
+		return nil, false
+	}
+	return entry, true
 }
 
-// we should make sure if it's the same control to prevent delete a new one
-func (cm *ControlManager) Del(runID string, ctl *Control) {
+// Add makes ctl the pending current generation and records the predecessor
+// finalization barrier it must wait for before activation.
+func (cm *ControlManager) Add(ctl *Control) error {
+	for {
+		// Never wait for a run gate while holding cm.mu.
+		cm.mu.RLock()
+		old := cm.ctlsByRunID[ctl.runID]
+		cm.mu.RUnlock()
+		if old != nil {
+			old.runMu.Lock()
+		}
+
+		cm.mu.Lock()
+		if cm.closed {
+			cm.mu.Unlock()
+			if old != nil {
+				old.runMu.Unlock()
+			}
+			return fmt.Errorf("control manager is closed")
+		}
+		if cm.ctlsByRunID[ctl.runID] != old {
+			cm.mu.Unlock()
+			if old != nil {
+				old.runMu.Unlock()
+			}
+			continue
+		}
+
+		id := ControlID(nextControlID.Add(1))
+		if err := ctl.admit(cm, id); err != nil {
+			cm.mu.Unlock()
+			if old != nil {
+				old.runMu.Unlock()
+			}
+			return err
+		}
+
+		runMu := &sync.Mutex{}
+		if old != nil {
+			runMu = old.runMu
+		}
+		entry := &controlEntry{ctl: ctl, id: id, runMu: runMu}
+		var (
+			oldCtl  *Control
+			barrier <-chan struct{}
+		)
+		if old != nil {
+			oldCtl = old.ctl
+			barrier = oldCtl.markReplaced()
+			ctl.setHandoffBarrier(barrier)
+			entry.registryOnline = old.registryOnline
+			entry.registryControlID = old.registryControlID
+		}
+		cm.ctlsByRunID[ctl.runID] = entry
+		cm.mu.Unlock()
+		if old != nil {
+			old.runMu.Unlock()
+		}
+
+		if oldCtl != nil {
+			oldCtl.Replaced(ctl)
+		}
+		return nil
+	}
+}
+
+// Activate registers ctl as online only if it is still the pending current
+// generation.
+func (cm *ControlManager) Activate(ctl *Control) (bool, error) {
+	entry, ok := cm.lockCurrentRun(ctl.runID, false)
+	if !ok {
+		return false, nil
+	}
+	defer entry.runMu.Unlock()
 	cm.mu.Lock()
 	defer cm.mu.Unlock()
-	if c, ok := cm.ctlsByRunID[runID]; ok && c == ctl {
-		delete(cm.ctlsByRunID, runID)
+
+	if cm.closed || cm.ctlsByRunID[ctl.runID] != entry || entry.ctl != ctl || entry.id != ctl.controlID {
+		return false, nil
 	}
+
+	ctl.lifecycleMu.Lock()
+	defer ctl.lifecycleMu.Unlock()
+	if ctl.state != controlStatePending {
+		return false, nil
+	}
+	if ctl.activated {
+		return true, nil
+	}
+
+	loginMsg := ctl.sessionCtx.LoginMsg
+	remoteAddr := ctl.sessionCtx.Conn.RemoteAddr().String()
+	if host, _, err := net.SplitHostPort(remoteAddr); err == nil {
+		remoteAddr = host
+	}
+	_, conflict := cm.registry.RegisterWithControlID(
+		loginMsg.User,
+		loginMsg.ClientID,
+		ctl.runID,
+		loginMsg.Hostname,
+		loginMsg.Version,
+		remoteAddr,
+		ctl.sessionCtx.WireProtocol,
+		uint64(entry.id),
+	)
+	if conflict {
+		return true, fmt.Errorf("client_id [%s] for user [%s] is already online", loginMsg.ClientID, loginMsg.User)
+	}
+
+	entry.registryOnline = true
+	entry.registryControlID = entry.id
+	ctl.activated = true
+	return true, nil
+}
+
+// completeLogin reserves ctl's current ownership with its run gate while the
+// bounded successful LoginResp write runs, then transitions it to running.
+// The callback must only perform that bounded write; it must not call back into
+// the control manager or the same control lifecycle.
+func (cm *ControlManager) completeLogin(ctl *Control, writeSuccess func() error) (bool, error) {
+	entry, ok := cm.lockCurrentRun(ctl.runID, false)
+	if !ok {
+		return false, nil
+	}
+	defer entry.runMu.Unlock()
+	if entry.ctl != ctl || entry.id != ctl.controlID {
+		return false, nil
+	}
+
+	ctl.lifecycleMu.Lock()
+	defer ctl.lifecycleMu.Unlock()
+	if ctl.state != controlStatePending || !ctl.activated {
+		return false, nil
+	}
+	if err := writeSuccess(); err != nil {
+		return false, err
+	}
+	if !ctl.startLocked() {
+		return false, nil
+	}
+	return true, nil
+}
+
+// Remove deletes and offlines ctl only if it is still the current generation.
+func (cm *ControlManager) Remove(ctl *Control) bool {
+	entry, ok := cm.lockCurrentRun(ctl.runID, true)
+	if !ok {
+		return false
+	}
+	defer entry.runMu.Unlock()
+	cm.mu.Lock()
+	defer cm.mu.Unlock()
+
+	if cm.ctlsByRunID[ctl.runID] != entry || entry.ctl != ctl || entry.id != ctl.controlID {
+		return false
+	}
+	delete(cm.ctlsByRunID, ctl.runID)
+	if entry.registryOnline {
+		cm.registry.MarkOfflineByRunIDAndControlID(ctl.runID, uint64(entry.registryControlID))
+	}
+	return true
 }
 
 func (cm *ControlManager) GetByID(runID string) (ctl *Control, ok bool) {
-	cm.mu.RLock()
-	defer cm.mu.RUnlock()
-	ctl, ok = cm.ctlsByRunID[runID]
-	return
+	entry, ok := cm.lockCurrentRun(runID, false)
+	if !ok {
+		return nil, false
+	}
+	defer entry.runMu.Unlock()
+	ctl = entry.ctl
+
+	ctl.lifecycleMu.Lock()
+	defer ctl.lifecycleMu.Unlock()
+	if ctl.state != controlStateRunning {
+		return nil, false
+	}
+	return ctl, true
+}
+
+// admitVisitorByRunID commits a visitor admission against the current running
+// control while its run and lifecycle ownership are held. The callback must
+// only perform the in-memory, buffered visitor admission.
+func (cm *ControlManager) admitVisitorByRunID(runID string, admit func(user string) error) (bool, error) {
+	entry, ok := cm.lockCurrentRun(runID, false)
+	if !ok {
+		return false, nil
+	}
+	defer entry.runMu.Unlock()
+	ctl := entry.ctl
+
+	ctl.lifecycleMu.Lock()
+	defer ctl.lifecycleMu.Unlock()
+	if ctl.state != controlStateRunning {
+		return false, nil
+	}
+	return true, admit(ctl.sessionCtx.LoginMsg.User)
+}
+
+// RegisterWorkConn transfers conn to ctl only if ctl is still the current
+// running generation. On error, ownership remains with the caller.
+func (cm *ControlManager) RegisterWorkConn(ctl *Control, conn *proxy.WorkConn) error {
+	entry, ok := cm.lockCurrentRun(ctl.runID, false)
+	if !ok {
+		cm.mu.RLock()
+		closed := cm.closed
+		cm.mu.RUnlock()
+		if closed {
+			return fmt.Errorf("control manager is closed")
+		}
+		return fmt.Errorf("client control for run id [%s] is no longer current", ctl.runID)
+	}
+	defer entry.runMu.Unlock()
+	if entry.ctl != ctl || entry.id != ctl.controlID {
+		return fmt.Errorf("client control for run id [%s] is no longer current", ctl.runID)
+	}
+
+	ctl.lifecycleMu.Lock()
+	defer ctl.lifecycleMu.Unlock()
+	if ctl.state != controlStateRunning {
+		return fmt.Errorf("client control for run id [%s] is not running", ctl.runID)
+	}
+
+	select {
+	case ctl.workConnCh <- conn:
+		ctl.xl.Debugf("new work connection registered")
+		return nil
+	default:
+		ctl.xl.Debugf("work connection pool is full, discarding")
+		return fmt.Errorf("work connection pool is full, discarding")
+	}
 }
 
 func (cm *ControlManager) Close() error {
 	cm.mu.Lock()
-	defer cm.mu.Unlock()
-	for _, ctl := range cm.ctlsByRunID {
-		ctl.Close()
+	cm.closed = true
+	ctls := make([]*Control, 0, len(cm.ctlsByRunID))
+	for _, entry := range cm.ctlsByRunID {
+		ctls = append(ctls, entry.ctl)
+	}
+	cm.mu.Unlock()
+
+	for _, ctl := range ctls {
+		cm.Remove(ctl)
+		_ = ctl.Close()
 	}
-	cm.ctlsByRunID = make(map[string]*Control)
 	return nil
 }
 
@@ -110,12 +367,20 @@ type SessionContext struct {
 	LoginMsg *msg.Login
 	// server configuration
 	ServerCfg *v1.ServerConfig
-	// client registry
-	ClientRegistry *registry.ClientRegistry
 	// negotiated wire protocol for this client session
 	WireProtocol string
 }
 
+type controlState uint8
+
+const (
+	controlStateCreated controlState = iota
+	controlStatePending
+	controlStateRunning
+	controlStateClosing
+	controlStateClosed
+)
+
 type Control struct {
 	// session context
 	sessionCtx *SessionContext
@@ -142,30 +407,42 @@ type Control struct {
 	// last time got the Ping message
 	lastPing atomic.Value
 
-	// A new run id will be generated when a new client login.
-	// If run id got from login message has same run id, it means it's the same client, so we can
-	// replace old controller instantly.
-	runID string
+	// runID never changes during the lifetime of a control. controlID is assigned
+	// once by ControlManager and distinguishes same-runID generations.
+	runID     string
+	controlID ControlID
+	manager   *ControlManager
+
+	lifecycleMu    sync.Mutex
+	state          controlState
+	activated      bool
+	handoffBarrier <-chan struct{}
+
+	interruptOnce sync.Once
+	interruptErr  error
 
 	mu sync.RWMutex
 
-	xl     *xlog.Logger
-	ctx    context.Context
-	doneCh chan struct{}
+	xl            *xlog.Logger
+	ctx           context.Context
+	doneCh        chan struct{}
+	serverMetrics metrics.ServerMetrics
 }
 
 func NewControl(ctx context.Context, sessionCtx *SessionContext) (*Control, error) {
 	poolCount := min(sessionCtx.LoginMsg.PoolCount, int(sessionCtx.ServerCfg.Transport.MaxPoolCount))
 	ctl := &Control{
-		sessionCtx:   sessionCtx,
-		workConnCh:   make(chan *proxy.WorkConn, poolCount+10),
-		proxies:      make(map[string]proxy.Proxy),
-		poolCount:    poolCount,
-		portsUsedNum: 0,
-		runID:        sessionCtx.LoginMsg.RunID,
-		xl:           xlog.FromContextSafe(ctx),
-		ctx:          ctx,
-		doneCh:       make(chan struct{}),
+		sessionCtx:    sessionCtx,
+		workConnCh:    make(chan *proxy.WorkConn, poolCount+10),
+		proxies:       make(map[string]proxy.Proxy),
+		poolCount:     poolCount,
+		portsUsedNum:  0,
+		runID:         sessionCtx.LoginMsg.RunID,
+		state:         controlStateCreated,
+		xl:            xlog.FromContextSafe(ctx),
+		ctx:           ctx,
+		doneCh:        make(chan struct{}),
+		serverMetrics: metrics.Server,
 	}
 	ctl.lastPing.Store(time.Now())
 
@@ -175,48 +452,121 @@ func NewControl(ctx context.Context, sessionCtx *SessionContext) (*Control, erro
 	return ctl, nil
 }
 
-// Start starts the control session workers after login succeeds.
-func (ctl *Control) Start() {
-	go func() {
-		for i := 0; i < ctl.poolCount; i++ {
-			// ignore error here, that means that this control is closed
-			_ = ctl.msgDispatcher.Send(&msg.ReqWorkConn{})
-		}
-	}()
-	go ctl.worker()
+func (ctl *Control) RunID() string {
+	return ctl.runID
 }
 
-func (ctl *Control) Close() error {
-	ctl.sessionCtx.Conn.Close()
+func (ctl *Control) ID() ControlID {
+	ctl.lifecycleMu.Lock()
+	defer ctl.lifecycleMu.Unlock()
+	return ctl.controlID
+}
+
+func (ctl *Control) admit(manager *ControlManager, id ControlID) error {
+	ctl.lifecycleMu.Lock()
+	defer ctl.lifecycleMu.Unlock()
+	if ctl.state != controlStateCreated {
+		return fmt.Errorf("control [%s] is not in created state", ctl.runID)
+	}
+	ctl.manager = manager
+	ctl.controlID = id
+	ctl.state = controlStatePending
 	return nil
 }
 
-func (ctl *Control) Replaced(newCtl *Control) {
-	xl := ctl.xl
-	xl.Infof("replaced by client [%s]", newCtl.runID)
-	ctl.runID = ""
-	ctl.sessionCtx.Conn.Close()
+func (ctl *Control) setHandoffBarrier(barrier <-chan struct{}) {
+	ctl.lifecycleMu.Lock()
+	ctl.handoffBarrier = barrier
+	ctl.lifecycleMu.Unlock()
 }
 
-func (ctl *Control) RegisterWorkConn(conn *proxy.WorkConn) error {
-	xl := ctl.xl
-	defer func() {
-		if err := recover(); err != nil {
-			xl.Errorf("panic error: %v", err)
-			xl.Errorf(string(debug.Stack()))
-		}
-	}()
-
-	select {
-	case ctl.workConnCh <- conn:
-		xl.Debugf("new work connection registered")
-		return nil
-	default:
-		xl.Debugf("work connection pool is full, discarding")
-		return fmt.Errorf("work connection pool is full, discarding")
+func (ctl *Control) WaitForHandoff() {
+	ctl.lifecycleMu.Lock()
+	barrier := ctl.handoffBarrier
+	ctl.lifecycleMu.Unlock()
+	if barrier != nil {
+		<-barrier
 	}
 }
 
+// Start starts the control session workers after login succeeds.
+func (ctl *Control) Start() bool {
+	ctl.lifecycleMu.Lock()
+	defer ctl.lifecycleMu.Unlock()
+	return ctl.startLocked()
+}
+
+func (ctl *Control) startLocked() bool {
+	if ctl.state != controlStatePending || !ctl.activated {
+		return false
+	}
+	ctl.state = controlStateRunning
+	go ctl.worker()
+	return true
+}
+
+func (ctl *Control) Close() error {
+	ctl.lifecycleMu.Lock()
+	switch ctl.state {
+	case controlStateCreated, controlStatePending:
+		ctl.state = controlStateClosing
+		ctl.finishLocked()
+	case controlStateRunning:
+		ctl.state = controlStateClosing
+	}
+	ctl.lifecycleMu.Unlock()
+	return ctl.interruptReadAndClose()
+}
+
+func (ctl *Control) Replaced(newCtl *Control) {
+	ctl.markReplaced()
+	ctl.xl.Infof("replaced by client [%s] (control ID %d)", newCtl.runID, newCtl.ID())
+	_ = ctl.interruptReadAndClose()
+}
+
+// markReplaced returns the transitive predecessor barrier. A pending control
+// has no worker, so it finishes immediately and passes its inherited barrier
+// to the replacement. A running control is finished only by its worker.
+func (ctl *Control) markReplaced() <-chan struct{} {
+	ctl.lifecycleMu.Lock()
+	defer ctl.lifecycleMu.Unlock()
+
+	switch ctl.state {
+	case controlStateCreated:
+		ctl.state = controlStateClosing
+		ctl.finishLocked()
+		return nil
+	case controlStatePending:
+		barrier := ctl.handoffBarrier
+		ctl.state = controlStateClosing
+		ctl.finishLocked()
+		return barrier
+	case controlStateRunning:
+		ctl.state = controlStateClosing
+		return ctl.doneCh
+	case controlStateClosing, controlStateClosed:
+		return ctl.doneCh
+	default:
+		return ctl.doneCh
+	}
+}
+
+func (ctl *Control) interruptReadAndClose() error {
+	ctl.interruptOnce.Do(func() {
+		_ = ctl.sessionCtx.Conn.SetReadDeadline(time.Now())
+		ctl.interruptErr = ctl.sessionCtx.Conn.Close()
+	})
+	return ctl.interruptErr
+}
+
+func (ctl *Control) finishLocked() {
+	if ctl.state == controlStateClosed {
+		return
+	}
+	ctl.state = controlStateClosed
+	close(ctl.doneCh)
+}
+
 // When frps get one user connection, we get one work connection from the pool and return it.
 // If no workConn available in the pool, send message to frpc to get one or more
 // and wait until it is available.
@@ -271,10 +621,10 @@ func (ctl *Control) heartbeatWorker() {
 	}
 
 	xl := ctl.xl
-	go wait.Until(func() {
+	wait.Until(func() {
 		if time.Since(ctl.lastPing.Load().(time.Time)) > time.Duration(ctl.sessionCtx.ServerCfg.Transport.HeartbeatTimeout)*time.Second {
 			xl.Warnf("heartbeat timeout")
-			ctl.sessionCtx.Conn.Close()
+			_ = ctl.Close()
 			return
 		}
 	}, time.Second, ctl.doneCh)
@@ -289,14 +639,14 @@ func (ctl *Control) loginUserInfo() plugin.UserInfo {
 	return plugin.UserInfo{
 		User:  ctl.sessionCtx.LoginMsg.User,
 		Metas: ctl.sessionCtx.LoginMsg.Metas,
-		RunID: ctl.sessionCtx.LoginMsg.RunID,
+		RunID: ctl.runID,
 	}
 }
 
 func (ctl *Control) closeProxy(pxy proxy.Proxy) {
 	pxy.Close()
 	ctl.sessionCtx.PxyManager.Del(pxy.GetName())
-	metrics.Server.CloseProxy(pxy.GetName(), pxy.GetConfigurer().GetBaseConfig().Type)
+	ctl.serverMetrics.CloseProxy(pxy.GetName(), pxy.GetConfigurer().GetBaseConfig().Type)
 
 	notifyContent := &plugin.CloseProxyContent{
 		User: ctl.loginUserInfo(),
@@ -311,12 +661,24 @@ func (ctl *Control) closeProxy(pxy proxy.Proxy) {
 
 func (ctl *Control) worker() {
 	xl := ctl.xl
+	ctl.serverMetrics.NewClient()
 
 	go ctl.heartbeatWorker()
 	go ctl.msgDispatcher.Run()
+	go func() {
+		for i := 0; i < ctl.poolCount; i++ {
+			// Ignore the error: it means this control is already closing.
+			_ = ctl.msgDispatcher.Send(&msg.ReqWorkConn{})
+		}
+	}()
 
 	<-ctl.msgDispatcher.Done()
-	ctl.sessionCtx.Conn.Close()
+	ctl.lifecycleMu.Lock()
+	if ctl.state == controlStateRunning {
+		ctl.state = controlStateClosing
+	}
+	ctl.lifecycleMu.Unlock()
+	_ = ctl.interruptReadAndClose()
 
 	ctl.mu.Lock()
 	close(ctl.workConnCh)
@@ -331,10 +693,14 @@ func (ctl *Control) worker() {
 		ctl.closeProxy(pxy)
 	}
 
-	metrics.Server.CloseClient()
-	ctl.sessionCtx.ClientRegistry.MarkOfflineByRunID(ctl.runID)
+	ctl.serverMetrics.CloseClient()
+	if ctl.manager != nil {
+		ctl.manager.Remove(ctl)
+	}
 	xl.Infof("client exit success")
-	close(ctl.doneCh)
+	ctl.lifecycleMu.Lock()
+	ctl.finishLocked()
+	ctl.lifecycleMu.Unlock()
 }
 
 func (ctl *Control) registerMsgHandlers() {
@@ -374,9 +740,9 @@ func (ctl *Control) handleNewProxy(m msg.Message) {
 		xl.Infof("new proxy [%s] type [%s] success", inMsg.ProxyName, inMsg.ProxyType)
 		clientID := ctl.sessionCtx.LoginMsg.ClientID
 		if clientID == "" {
-			clientID = ctl.sessionCtx.LoginMsg.RunID
+			clientID = ctl.runID
 		}
-		metrics.Server.NewProxy(inMsg.ProxyName, inMsg.ProxyType, ctl.sessionCtx.LoginMsg.User, clientID)
+		ctl.serverMetrics.NewProxy(inMsg.ProxyName, inMsg.ProxyType, ctl.sessionCtx.LoginMsg.User, clientID)
 	}
 	_ = ctl.msgDispatcher.Send(resp)
 }

server/control_test.go

@@ -0,0 +1,543 @@
+// Copyright 2026 The frp Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package server
+
+import (
+	"context"
+	"errors"
+	"net"
+	"os"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/fatedier/frp/pkg/auth"
+	v1 "github.com/fatedier/frp/pkg/config/v1"
+	"github.com/fatedier/frp/pkg/msg"
+	plugin "github.com/fatedier/frp/pkg/plugin/server"
+	"github.com/fatedier/frp/server/controller"
+	"github.com/fatedier/frp/server/proxy"
+	"github.com/fatedier/frp/server/registry"
+)
+
+func TestControlPendingReplacementFinishesWithoutStarting(t *testing.T) {
+	clientRegistry := registry.NewClientRegistry()
+	manager := NewControlManager(clientRegistry)
+	metrics := newCountingServerMetrics()
+	oldCtl, oldConn := newLifecycleTestControl(t, "same-run", "client", metrics)
+	newCtl, _ := newLifecycleTestControl(t, "same-run", "client", metrics)
+
+	mustAddAndActivate(t, manager, oldCtl)
+
+	err := manager.Add(newCtl)
+	require.NoError(t, err)
+	waitForControlDone(t, oldCtl)
+	require.False(t, oldCtl.Start())
+	require.Equal(t, []string{"deadline", "close"}, oldConn.eventsSnapshot())
+	require.Equal(t, int64(0), metrics.newClients())
+	require.Equal(t, int64(0), metrics.closedClients())
+}
+
+func TestControlRunningReplacementFinishesInWorker(t *testing.T) {
+	clientRegistry := registry.NewClientRegistry()
+	manager := NewControlManager(clientRegistry)
+	metrics := newCountingServerMetrics()
+	oldCtl, oldConn := newLifecycleTestControl(t, "same-run", "client", metrics)
+	newCtl, _ := newLifecycleTestControl(t, "same-run", "client", metrics)
+
+	mustAddAndActivate(t, manager, oldCtl)
+	require.True(t, oldCtl.Start())
+	waitForSignal(t, oldConn.readStarted, "control reader to start")
+
+	err := manager.Add(newCtl)
+	require.NoError(t, err)
+	waitForControlDone(t, oldCtl)
+	require.Equal(t, []string{"deadline", "close"}, oldConn.eventsSnapshot())
+	require.Equal(t, int64(1), metrics.newClients())
+	require.Equal(t, int64(1), metrics.closedClients())
+
+	_, ok := manager.GetByID("same-run")
+	require.False(t, ok)
+	require.Same(t, newCtl, currentControlForTest(manager, "same-run"))
+	info, ok := clientRegistry.GetByKey("client")
+	require.True(t, ok)
+	require.True(t, info.Online)
+	require.Equal(t, uint64(oldCtl.ID()), info.ControlID)
+
+	active, err := manager.Activate(newCtl)
+	require.NoError(t, err)
+	require.True(t, active)
+	_, ok = manager.GetByID("same-run")
+	require.False(t, ok)
+	info, ok = clientRegistry.GetByKey("client")
+	require.True(t, ok)
+	require.Equal(t, uint64(newCtl.ID()), info.ControlID)
+}
+
+func TestControlClosePendingAndRunning(t *testing.T) {
+	t.Run("pending", func(t *testing.T) {
+		manager := NewControlManager(registry.NewClientRegistry())
+		metrics := newCountingServerMetrics()
+		ctl, conn := newLifecycleTestControl(t, "pending", "pending", metrics)
+		err := manager.Add(ctl)
+		require.NoError(t, err)
+
+		require.NoError(t, ctl.Close())
+		waitForControlDone(t, ctl)
+		require.Equal(t, []string{"deadline", "close"}, conn.eventsSnapshot())
+		require.Equal(t, int64(0), metrics.newClients())
+		require.Equal(t, int64(0), metrics.closedClients())
+	})
+
+	t.Run("running", func(t *testing.T) {
+		manager := NewControlManager(registry.NewClientRegistry())
+		metrics := newCountingServerMetrics()
+		ctl, conn := newLifecycleTestControl(t, "running", "running", metrics)
+		mustAddAndActivate(t, manager, ctl)
+		require.True(t, ctl.Start())
+		waitForSignal(t, conn.readStarted, "control reader to start")
+
+		require.NoError(t, ctl.Close())
+		waitForControlDone(t, ctl)
+		require.Equal(t, []string{"deadline", "close"}, conn.eventsSnapshot())
+		require.Equal(t, int64(1), metrics.newClients())
+		require.Equal(t, int64(1), metrics.closedClients())
+	})
+}
+
+func TestControlCloseAndReplacedAreIdempotent(t *testing.T) {
+	manager := NewControlManager(registry.NewClientRegistry())
+	metrics := newCountingServerMetrics()
+	ctl, conn := newLifecycleTestControl(t, "same-run", "client", metrics)
+	replacement, _ := newLifecycleTestControl(t, "same-run", "client", metrics)
+
+	err := manager.Add(ctl)
+	require.NoError(t, err)
+	err = manager.Add(replacement)
+	require.NoError(t, err)
+	require.NoError(t, ctl.Close())
+	ctl.Replaced(replacement)
+	require.NoError(t, ctl.Close())
+	waitForControlDone(t, ctl)
+
+	require.Equal(t, []string{"deadline", "close"}, conn.eventsSnapshot())
+	require.Equal(t, int64(0), metrics.newClients())
+	require.Equal(t, int64(0), metrics.closedClients())
+}
+
+func TestControlHeartbeatTimeoutInterruptsRead(t *testing.T) {
+	manager := NewControlManager(registry.NewClientRegistry())
+	metrics := newCountingServerMetrics()
+	ctl, conn := newLifecycleTestControl(t, "heartbeat", "heartbeat", metrics)
+	ctl.sessionCtx.ServerCfg.Transport.HeartbeatTimeout = 1
+	ctl.lastPing.Store(time.Now().Add(-2 * time.Second))
+
+	mustAddAndActivate(t, manager, ctl)
+	require.True(t, ctl.Start())
+	waitForSignal(t, conn.readStarted, "control reader to start")
+	waitForControlDone(t, ctl)
+
+	require.Equal(t, []string{"deadline", "close"}, conn.eventsSnapshot())
+	require.Equal(t, int64(1), metrics.newClients())
+	require.Equal(t, int64(1), metrics.closedClients())
+}
+
+func TestControlStartReplacementRacePairsMetrics(t *testing.T) {
+	for range 100 {
+		clientRegistry := registry.NewClientRegistry()
+		manager := NewControlManager(clientRegistry)
+		metrics := newCountingServerMetrics()
+		ctl, _ := newLifecycleTestControl(t, "same-run", "client", metrics)
+		replacement, _ := newLifecycleTestControl(t, "same-run", "client", metrics)
+
+		mustAddAndActivate(t, manager, ctl)
+
+		startGate := make(chan struct{})
+		startedCh := make(chan bool, 1)
+		addErrCh := make(chan error, 1)
+		go func() {
+			<-startGate
+			startedCh <- ctl.Start()
+		}()
+		go func() {
+			<-startGate
+			addErr := manager.Add(replacement)
+			addErrCh <- addErr
+		}()
+		close(startGate)
+
+		started := <-startedCh
+		require.NoError(t, <-addErrCh)
+		waitForControlDone(t, ctl)
+		if started {
+			require.Equal(t, int64(1), metrics.newClients())
+			require.Equal(t, int64(1), metrics.closedClients())
+		} else {
+			require.Equal(t, int64(0), metrics.newClients())
+			require.Equal(t, int64(0), metrics.closedClients())
+		}
+	}
+}
+
+func TestControlManagerRejectsStaleActivateAndRemove(t *testing.T) {
+	clientRegistry := registry.NewClientRegistry()
+	manager := NewControlManager(clientRegistry)
+	metrics := newCountingServerMetrics()
+	oldCtl, _ := newLifecycleTestControl(t, "same-run", "client", metrics)
+	newCtl, _ := newLifecycleTestControl(t, "same-run", "client", metrics)
+
+	mustAddAndActivate(t, manager, oldCtl)
+	err := manager.Add(newCtl)
+	require.NoError(t, err)
+	require.Greater(t, uint64(newCtl.ID()), uint64(oldCtl.ID()))
+
+	active, err := manager.Activate(oldCtl)
+	require.NoError(t, err)
+	require.False(t, active)
+	require.False(t, manager.Remove(oldCtl))
+
+	_, ok := manager.GetByID("same-run")
+	require.False(t, ok)
+	require.Same(t, newCtl, currentControlForTest(manager, "same-run"))
+	info, ok := clientRegistry.GetByKey("client")
+	require.True(t, ok)
+	require.True(t, info.Online)
+	require.Equal(t, uint64(oldCtl.ID()), info.ControlID)
+
+	active, err = manager.Activate(newCtl)
+	require.NoError(t, err)
+	require.True(t, active)
+	info, ok = clientRegistry.GetByKey("client")
+	require.True(t, ok)
+	require.True(t, info.Online)
+	require.Equal(t, uint64(newCtl.ID()), info.ControlID)
+}
+
+func TestControlManagerPreservesClientIDConflict(t *testing.T) {
+	clientRegistry := registry.NewClientRegistry()
+	manager := NewControlManager(clientRegistry)
+	metrics := newCountingServerMetrics()
+	first, _ := newLifecycleTestControl(t, "run-one", "shared-client", metrics)
+	conflicting, _ := newLifecycleTestControl(t, "run-two", "shared-client", metrics)
+
+	mustAddAndActivate(t, manager, first)
+	err := manager.Add(conflicting)
+	require.NoError(t, err)
+	active, err := manager.Activate(conflicting)
+	require.True(t, active)
+	require.ErrorContains(t, err, "already online")
+
+	require.True(t, manager.Remove(conflicting))
+	info, ok := clientRegistry.GetByKey("shared-client")
+	require.True(t, ok)
+	require.True(t, info.Online)
+	require.Equal(t, "run-one", info.RunID)
+}
+
+func TestControlManagerFailedLoginWriteReleasesRunWithoutStarting(t *testing.T) {
+	clientRegistry := registry.NewClientRegistry()
+	manager := NewControlManager(clientRegistry)
+	metrics := newCountingServerMetrics()
+	ctl, _ := newLifecycleTestControl(t, "same-run", "client", metrics)
+	replacement, _ := newLifecycleTestControl(t, "same-run", "client", metrics)
+
+	mustAddAndActivate(t, manager, ctl)
+
+	writeErr := errors.New("write failed")
+	committed, err := manager.completeLogin(ctl, func() error { return writeErr })
+	require.ErrorIs(t, err, writeErr)
+	require.False(t, committed)
+
+	err = manager.Add(replacement)
+	require.NoError(t, err)
+	waitForControlDone(t, ctl)
+	require.Same(t, replacement, currentControlForTest(manager, "same-run"))
+	require.Equal(t, int64(0), metrics.newClients())
+	require.Equal(t, int64(0), metrics.closedClients())
+	require.True(t, manager.Remove(replacement))
+	info, ok := clientRegistry.GetByKey("client")
+	require.True(t, ok)
+	require.False(t, info.Online)
+	require.Empty(t, info.RunID)
+	require.Zero(t, info.ControlID)
+	require.False(t, info.DisconnectedAt.IsZero())
+	require.NoError(t, replacement.Close())
+}
+
+func TestControlManagerCloseWaitsForInFlightLoginRun(t *testing.T) {
+	clientRegistry := registry.NewClientRegistry()
+	manager := NewControlManager(clientRegistry)
+	metrics := newCountingServerMetrics()
+	ctl, _ := newLifecycleTestControl(t, "same-run", "client", metrics)
+
+	mustAddAndActivate(t, manager, ctl)
+
+	writeEntered := make(chan struct{})
+	resumeWrite := make(chan struct{})
+	loginDone := make(chan struct {
+		committed bool
+		err       error
+	}, 1)
+	go func() {
+		committed, loginErr := manager.completeLogin(ctl, func() error {
+			close(writeEntered)
+			<-resumeWrite
+			return nil
+		})
+		loginDone <- struct {
+			committed bool
+			err       error
+		}{committed: committed, err: loginErr}
+	}()
+	waitForSignal(t, writeEntered, "LoginResp write")
+
+	closeDone := make(chan error, 1)
+	go func() { closeDone <- manager.Close() }()
+	waitForManagerClosed(t, manager)
+	select {
+	case err := <-closeDone:
+		t.Fatalf("manager close completed during LoginResp write: %v", err)
+	default:
+	}
+
+	close(resumeWrite)
+	result := <-loginDone
+	require.NoError(t, result.err)
+	require.True(t, result.committed)
+	require.NoError(t, <-closeDone)
+	waitForControlDone(t, ctl)
+	require.Nil(t, currentControlForTest(manager, "same-run"))
+	require.Equal(t, int64(1), metrics.newClients())
+	require.Equal(t, int64(1), metrics.closedClients())
+	info, ok := clientRegistry.GetByKey("client")
+	require.True(t, ok)
+	require.False(t, info.Online)
+}
+
+func newLifecycleTestControl(
+	t *testing.T,
+	runID string,
+	clientID string,
+	serverMetrics *countingServerMetrics,
+) (*Control, *deadlineReadConn) {
+	t.Helper()
+	conn := newDeadlineReadConn()
+	msgConn := msg.NewConn(conn, msg.NewV1ReadWriter(conn))
+	ctl, err := NewControl(context.Background(), &SessionContext{
+		RC:            &controller.ResourceController{},
+		PxyManager:    proxy.NewManager(),
+		PluginManager: plugin.NewManager(),
+		AuthVerifier:  auth.AlwaysPassVerifier,
+		Conn:          msgConn,
+		LoginMsg: &msg.Login{
+			RunID:    runID,
+			ClientID: clientID,
+		},
+		ServerCfg: &v1.ServerConfig{},
+	})
+	require.NoError(t, err)
+	ctl.serverMetrics = serverMetrics
+	t.Cleanup(func() { _ = ctl.Close() })
+	return ctl, conn
+}
+
+func mustAddAndActivate(t *testing.T, manager *ControlManager, ctl *Control) {
+	t.Helper()
+	require.NoError(t, manager.Add(ctl))
+	active, err := manager.Activate(ctl)
+	require.NoError(t, err)
+	require.True(t, active)
+}
+
+func waitForControlDone(t *testing.T, ctl *Control) {
+	t.Helper()
+	done := make(chan struct{})
+	go func() {
+		ctl.WaitClosed()
+		close(done)
+	}()
+	waitForSignal(t, done, "control to finish")
+}
+
+func currentControlForTest(manager *ControlManager, runID string) *Control {
+	manager.mu.RLock()
+	defer manager.mu.RUnlock()
+	entry := manager.ctlsByRunID[runID]
+	if entry == nil {
+		return nil
+	}
+	return entry.ctl
+}
+
+func currentRunGateForTest(manager *ControlManager, runID string) *sync.Mutex {
+	manager.mu.RLock()
+	defer manager.mu.RUnlock()
+	entry := manager.ctlsByRunID[runID]
+	if entry == nil {
+		return nil
+	}
+	return entry.runMu
+}
+
+func waitForManagerClosed(t *testing.T, manager *ControlManager) {
+	t.Helper()
+	deadline := time.Now().Add(3 * time.Second)
+	for time.Now().Before(deadline) {
+		manager.mu.RLock()
+		closed := manager.closed
+		manager.mu.RUnlock()
+		if closed {
+			return
+		}
+	}
+	t.Fatal("timed out waiting for control manager to close")
+}
+
+func waitForSignal(t *testing.T, ch <-chan struct{}, description string) {
+	t.Helper()
+	select {
+	case <-ch:
+	case <-time.After(3 * time.Second):
+		t.Fatalf("timed out waiting for %s", description)
+	}
+}
+
+type deadlineReadConn struct {
+	readStarted chan struct{}
+	unblockRead chan struct{}
+
+	readOnce     sync.Once
+	unblockOnce  sync.Once
+	deadlineOnce sync.Once
+	closeOnce    sync.Once
+
+	eventsMu sync.Mutex
+	events   []string
+}
+
+func newDeadlineReadConn() *deadlineReadConn {
+	return &deadlineReadConn{
+		readStarted: make(chan struct{}),
+		unblockRead: make(chan struct{}),
+	}
+}
+
+func (c *deadlineReadConn) Read([]byte) (int, error) {
+	c.readOnce.Do(func() { close(c.readStarted) })
+	<-c.unblockRead
+	return 0, os.ErrDeadlineExceeded
+}
+
+func (*deadlineReadConn) Write(p []byte) (int, error) { return len(p), nil }
+
+func (c *deadlineReadConn) Close() error {
+	c.closeOnce.Do(func() {
+		c.recordEvent("close")
+		c.unblockOnce.Do(func() { close(c.unblockRead) })
+	})
+	return nil
+}
+
+func (*deadlineReadConn) LocalAddr() net.Addr  { return lifecycleTestAddr("local") }
+func (*deadlineReadConn) RemoteAddr() net.Addr { return lifecycleTestAddr("remote") }
+
+func (c *deadlineReadConn) SetDeadline(deadline time.Time) error {
+	if err := c.SetReadDeadline(deadline); err != nil {
+		return err
+	}
+	return c.SetWriteDeadline(deadline)
+}
+
+func (c *deadlineReadConn) SetReadDeadline(deadline time.Time) error {
+	if deadline.IsZero() {
+		return nil
+	}
+	c.deadlineOnce.Do(func() {
+		c.recordEvent("deadline")
+		c.unblockOnce.Do(func() { close(c.unblockRead) })
+	})
+	return nil
+}
+
+func (*deadlineReadConn) SetWriteDeadline(time.Time) error { return nil }
+
+func (c *deadlineReadConn) recordEvent(event string) {
+	c.eventsMu.Lock()
+	c.events = append(c.events, event)
+	c.eventsMu.Unlock()
+}
+
+func (c *deadlineReadConn) eventsSnapshot() []string {
+	c.eventsMu.Lock()
+	defer c.eventsMu.Unlock()
+	return append([]string(nil), c.events...)
+}
+
+type lifecycleTestAddr string
+
+func (a lifecycleTestAddr) Network() string { return string(a) }
+func (a lifecycleTestAddr) String() string  { return string(a) }
+
+type countingServerMetrics struct {
+	mu          sync.Mutex
+	newCount    int64
+	closeCount  int64
+	closeEnter  chan struct{}
+	closeResume chan struct{}
+	closeOnce   sync.Once
+}
+
+func newCountingServerMetrics() *countingServerMetrics {
+	return &countingServerMetrics{}
+}
+
+func (m *countingServerMetrics) NewClient() {
+	m.mu.Lock()
+	m.newCount++
+	m.mu.Unlock()
+}
+
+func (m *countingServerMetrics) CloseClient() {
+	m.mu.Lock()
+	m.closeCount++
+	closeEnter := m.closeEnter
+	closeResume := m.closeResume
+	m.mu.Unlock()
+	if closeEnter != nil {
+		m.closeOnce.Do(func() { close(closeEnter) })
+		<-closeResume
+	}
+}
+
+func (*countingServerMetrics) NewProxy(string, string, string, string) {}
+func (*countingServerMetrics) CloseProxy(string, string)               {}
+func (*countingServerMetrics) OpenConnection(string, string)           {}
+func (*countingServerMetrics) CloseConnection(string, string)          {}
+func (*countingServerMetrics) AddTrafficIn(string, string, int64)      {}
+func (*countingServerMetrics) AddTrafficOut(string, string, int64)     {}
+
+func (m *countingServerMetrics) newClients() int64 {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return m.newCount
+}
+
+func (m *countingServerMetrics) closedClients() int64 {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return m.closeCount
+}

server/http/controller_v2.go

@@ -17,12 +17,14 @@ package http
 import (
 	"cmp"
 	"fmt"
+	"maps"
 	"math"
 	"net/http"
 	"net/url"
 	"slices"
 	"strconv"
 	"strings"
+	"time"
 
 	v1 "github.com/fatedier/frp/pkg/config/v1"
 	"github.com/fatedier/frp/pkg/metrics/mem"
@@ -37,6 +39,9 @@ const (
 	maxV2PageSize     = 200
 
 	v2SystemPruneTypeOfflineProxies = "offline_proxies"
+	v2ProxyTrafficDefaultDays       = 7
+	v2ProxyTrafficUnit              = "bytes"
+	v2ProxyTrafficGranularity       = "day"
 )
 
 var apiV2ProxyTypes = []string{
@@ -190,15 +195,10 @@ func (c *Controller) APIV2ClientList(ctx *httppkg.Context) (any, error) {
 
 // /api/v2/clients/{key}
 func (c *Controller) APIV2ClientDetail(ctx *httppkg.Context) (any, error) {
-	key := ctx.Param("key")
-	if key == "" {
-		return nil, fmt.Errorf("missing client key")
-	}
-	decodedKey, err := url.PathUnescape(key)
+	key, err := decodeV2PathParam(ctx, "key", "client key")
 	if err != nil {
-		return nil, fmt.Errorf("invalid client key %q: %w", key, err)
+		return nil, err
 	}
-	key = decodedKey
 
 	if c.clientRegistry == nil {
 		return nil, fmt.Errorf("client registry unavailable")
@@ -256,7 +256,7 @@ func (c *Controller) APIV2ProxyList(ctx *httppkg.Context) (any, error) {
 	}
 
 	slices.SortFunc(items, func(a, b model.V2ProxyResp) int {
-		if v := cmp.Compare(a.Type, b.Type); v != 0 {
+		if v := cmp.Compare(a.Spec.Type, b.Spec.Type); v != 0 {
 			return v
 		}
 		return cmp.Compare(a.Name, b.Name)
@@ -267,9 +267,9 @@ func (c *Controller) APIV2ProxyList(ctx *httppkg.Context) (any, error) {
 
 // /api/v2/proxies/{name}
 func (c *Controller) APIV2ProxyDetail(ctx *httppkg.Context) (any, error) {
-	name := ctx.Param("name")
-	if name == "" {
-		return nil, fmt.Errorf("missing proxy name")
+	name, err := decodeV2PathParam(ctx, "name", "proxy name")
+	if err != nil {
+		return nil, err
 	}
 
 	ps := mem.StatsCollector.GetProxyByName(name)
@@ -279,6 +279,33 @@ func (c *Controller) APIV2ProxyDetail(ctx *httppkg.Context) (any, error) {
 	return c.buildV2ProxyResp(ps), nil
 }
 
+// /api/v2/proxies/{name}/traffic
+func (c *Controller) APIV2ProxyTraffic(ctx *httppkg.Context) (any, error) {
+	name, err := decodeV2PathParam(ctx, "name", "proxy name")
+	if err != nil {
+		return nil, err
+	}
+
+	proxyTrafficInfo := mem.StatsCollector.GetProxyTraffic(name)
+	if proxyTrafficInfo == nil {
+		return nil, httppkg.NewError(http.StatusNotFound, "no proxy info found")
+	}
+
+	return buildV2ProxyTrafficResp(name, proxyTrafficInfo, time.Now()), nil
+}
+
+func decodeV2PathParam(ctx *httppkg.Context, key string, label string) (string, error) {
+	raw := ctx.Param(key)
+	if raw == "" {
+		return "", fmt.Errorf("missing %s", label)
+	}
+	decoded, err := url.PathUnescape(raw)
+	if err != nil {
+		return "", httppkg.NewError(http.StatusBadRequest, fmt.Sprintf("invalid %s", label))
+	}
+	return decoded, nil
+}
+
 func getOrCreateV2User(items map[string]*model.V2UserResp, user string) *model.V2UserResp {
 	item, ok := items[user]
 	if !ok {
@@ -409,26 +436,36 @@ func matchV2ClientQuery(item model.ClientInfoResp, q string) bool {
 func matchV2ProxyQuery(item model.V2ProxyResp, q string) bool {
 	values := []string{
 		item.Name,
-		item.Type,
+		item.Spec.Type,
 		item.User,
 		item.ClientID,
 		item.Status.State,
 	}
 
-	switch spec := item.Spec.(type) {
-	case *model.TCPOutConf:
-		values = append(values, strconv.Itoa(spec.RemotePort))
-	case *model.UDPOutConf:
-		values = append(values, strconv.Itoa(spec.RemotePort))
-	case *model.HTTPOutConf:
-		values = append(values, spec.CustomDomains...)
-		values = append(values, spec.SubDomain)
-	case *model.HTTPSOutConf:
-		values = append(values, spec.CustomDomains...)
-		values = append(values, spec.SubDomain)
-	case *model.TCPMuxOutConf:
-		values = append(values, spec.CustomDomains...)
-		values = append(values, spec.SubDomain)
+	switch item.Spec.Type {
+	case string(v1.ProxyTypeTCP):
+		if item.Spec.TCP != nil && item.Spec.TCP.RemotePort != nil {
+			values = append(values, strconv.Itoa(*item.Spec.TCP.RemotePort))
+		}
+	case string(v1.ProxyTypeUDP):
+		if item.Spec.UDP != nil && item.Spec.UDP.RemotePort != nil {
+			values = append(values, strconv.Itoa(*item.Spec.UDP.RemotePort))
+		}
+	case string(v1.ProxyTypeHTTP):
+		if item.Spec.HTTP != nil {
+			values = append(values, item.Spec.HTTP.CustomDomains...)
+			values = append(values, item.Spec.HTTP.Subdomain)
+		}
+	case string(v1.ProxyTypeHTTPS):
+		if item.Spec.HTTPS != nil {
+			values = append(values, item.Spec.HTTPS.CustomDomains...)
+			values = append(values, item.Spec.HTTPS.Subdomain)
+		}
+	case string(v1.ProxyTypeTCPMUX):
+		if item.Spec.TCPMux != nil {
+			values = append(values, item.Spec.TCPMux.CustomDomains...)
+			values = append(values, item.Spec.TCPMux.Subdomain)
+		}
 	}
 
 	return containsV2Query(q, values...)
@@ -455,6 +492,31 @@ func (c *Controller) listV2ProxyStats(proxyType string) []*mem.ProxyStats {
 	return items
 }
 
+func buildV2ProxyTrafficResp(name string, traffic *mem.ProxyTrafficInfo, now time.Time) model.V2ProxyTrafficResp {
+	history := make([]model.V2ProxyTrafficPointResp, 0, v2ProxyTrafficDefaultDays)
+	for age := v2ProxyTrafficDefaultDays - 1; age >= 0; age-- {
+		history = append(history, model.V2ProxyTrafficPointResp{
+			Date:       now.AddDate(0, 0, -age).Format(time.DateOnly),
+			TrafficIn:  v2TrafficValueAt(traffic.TrafficIn, age),
+			TrafficOut: v2TrafficValueAt(traffic.TrafficOut, age),
+		})
+	}
+
+	return model.V2ProxyTrafficResp{
+		Name:        name,
+		Unit:        v2ProxyTrafficUnit,
+		Granularity: v2ProxyTrafficGranularity,
+		History:     history,
+	}
+}
+
+func v2TrafficValueAt(values []int64, todayFirstIndex int) int64 {
+	if todayFirstIndex >= len(values) {
+		return 0
+	}
+	return values[todayFirstIndex]
+}
+
 func (c *Controller) buildV2ClientStatus(info registry.ClientInfo) model.V2ClientStatusResp {
 	status := model.V2ClientStatusResp{State: "offline"}
 	if info.Online {
@@ -475,27 +537,111 @@ func (c *Controller) buildV2ClientStatus(info registry.ClientInfo) model.V2Clien
 
 func (c *Controller) buildV2ProxyResp(ps *mem.ProxyStats) model.V2ProxyResp {
 	state := "offline"
-	var spec any
+	var cfg v1.ProxyConfigurer
 	if c.pxyManager != nil {
 		if pxy, ok := c.pxyManager.GetByName(ps.Name); ok {
 			state = "online"
-			spec = getConfFromConfigurer(pxy.GetConfigurer())
+			cfg = pxy.GetConfigurer()
 		}
 	}
 
 	return model.V2ProxyResp{
 		Name:     ps.Name,
-		Type:     ps.Type,
 		User:     ps.User,
 		ClientID: ps.ClientID,
-		Spec:     spec,
+		Spec:     buildV2ProxySpec(ps.Type, cfg),
 		Status: model.V2ProxyStatusResp{
 			State:           state,
 			TodayTrafficIn:  ps.TodayTrafficIn,
 			TodayTrafficOut: ps.TodayTrafficOut,
 			CurConns:        ps.CurConns,
-			LastStartTime:   ps.LastStartTime,
-			LastCloseTime:   ps.LastCloseTime,
+			LastStartAt:     ps.LastStartAt,
+			LastCloseAt:     ps.LastCloseAt,
+		},
+	}
+}
+
+func buildV2ProxySpec(proxyType string, cfg v1.ProxyConfigurer) model.V2ProxySpec {
+	spec := model.V2ProxySpec{Type: proxyType}
+
+	switch proxyType {
+	case string(v1.ProxyTypeTCP):
+		block := &model.V2TCPProxySpec{}
+		if c, ok := cfg.(*v1.TCPProxyConfig); ok {
+			block.V2ProxyBaseSpec = buildV2ProxyBaseSpec(c.GetBaseConfig())
+			block.RemotePort = &c.RemotePort
+		}
+		spec.TCP = block
+	case string(v1.ProxyTypeUDP):
+		block := &model.V2UDPProxySpec{}
+		if c, ok := cfg.(*v1.UDPProxyConfig); ok {
+			block.V2ProxyBaseSpec = buildV2ProxyBaseSpec(c.GetBaseConfig())
+			block.RemotePort = &c.RemotePort
+		}
+		spec.UDP = block
+	case string(v1.ProxyTypeHTTP):
+		block := &model.V2HTTPProxySpec{}
+		if c, ok := cfg.(*v1.HTTPProxyConfig); ok {
+			block.V2ProxyBaseSpec = buildV2ProxyBaseSpec(c.GetBaseConfig())
+			block.CustomDomains = slices.Clone(c.CustomDomains)
+			block.Subdomain = c.SubDomain
+			block.Locations = slices.Clone(c.Locations)
+			block.HostHeaderRewrite = c.HostHeaderRewrite
+		}
+		spec.HTTP = block
+	case string(v1.ProxyTypeHTTPS):
+		block := &model.V2HTTPSProxySpec{}
+		if c, ok := cfg.(*v1.HTTPSProxyConfig); ok {
+			block.V2ProxyBaseSpec = buildV2ProxyBaseSpec(c.GetBaseConfig())
+			block.CustomDomains = slices.Clone(c.CustomDomains)
+			block.Subdomain = c.SubDomain
+		}
+		spec.HTTPS = block
+	case string(v1.ProxyTypeTCPMUX):
+		block := &model.V2TCPMuxProxySpec{}
+		if c, ok := cfg.(*v1.TCPMuxProxyConfig); ok {
+			block.V2ProxyBaseSpec = buildV2ProxyBaseSpec(c.GetBaseConfig())
+			block.CustomDomains = slices.Clone(c.CustomDomains)
+			block.Subdomain = c.SubDomain
+			block.Multiplexer = c.Multiplexer
+			block.RouteByHTTPUser = c.RouteByHTTPUser
+		}
+		spec.TCPMux = block
+	case string(v1.ProxyTypeSTCP):
+		block := &model.V2STCPProxySpec{}
+		if c, ok := cfg.(*v1.STCPProxyConfig); ok {
+			block.V2ProxyBaseSpec = buildV2ProxyBaseSpec(c.GetBaseConfig())
+		}
+		spec.STCP = block
+	case string(v1.ProxyTypeSUDP):
+		block := &model.V2SUDPProxySpec{}
+		if c, ok := cfg.(*v1.SUDPProxyConfig); ok {
+			block.V2ProxyBaseSpec = buildV2ProxyBaseSpec(c.GetBaseConfig())
+		}
+		spec.SUDP = block
+	case string(v1.ProxyTypeXTCP):
+		block := &model.V2XTCPProxySpec{}
+		if c, ok := cfg.(*v1.XTCPProxyConfig); ok {
+			block.V2ProxyBaseSpec = buildV2ProxyBaseSpec(c.GetBaseConfig())
+		}
+		spec.XTCP = block
+	}
+
+	return spec
+}
+
+func buildV2ProxyBaseSpec(base *v1.ProxyBaseConfig) model.V2ProxyBaseSpec {
+	return model.V2ProxyBaseSpec{
+		Annotations: maps.Clone(base.Annotations),
+		Metadatas:   maps.Clone(base.Metadatas),
+		Transport: &model.V2ProxyTransportSpec{
+			UseEncryption:      base.Transport.UseEncryption,
+			UseCompression:     base.Transport.UseCompression,
+			BandwidthLimit:     base.Transport.BandwidthLimit.String(),
+			BandwidthLimitMode: base.Transport.BandwidthLimitMode,
+		},
+		LoadBalancer: &model.V2ProxyLoadBalancerSpec{
+			Group: base.LoadBalancer.Group,
 		},
 	}
 }

server/http/controller_v2_proxy_spec_test.go

@@ -0,0 +1,393 @@
+// Copyright 2026 The frp Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package http
+
+import (
+	"encoding/json"
+	"strings"
+	"testing"
+
+	configtypes "github.com/fatedier/frp/pkg/config/types"
+	v1 "github.com/fatedier/frp/pkg/config/v1"
+	"github.com/fatedier/frp/pkg/metrics/mem"
+	"github.com/fatedier/frp/server/http/model"
+)
+
+func TestBuildV2ProxySpecAllTypesAndRedaction(t *testing.T) {
+	tests := []struct {
+		proxyType string
+		cfg       v1.ProxyConfigurer
+		blockKeys []string
+	}{
+		{
+			proxyType: "tcp",
+			cfg: &v1.TCPProxyConfig{
+				ProxyBaseConfig: newV2ProxyTestBaseConfig(t, "tcp"),
+				RemotePort:      6000,
+			},
+			blockKeys: []string{"annotations", "loadBalancer", "metadatas", "remotePort", "transport"},
+		},
+		{
+			proxyType: "udp",
+			cfg: &v1.UDPProxyConfig{
+				ProxyBaseConfig: newV2ProxyTestBaseConfig(t, "udp"),
+				RemotePort:      7000,
+			},
+			blockKeys: []string{"annotations", "loadBalancer", "metadatas", "remotePort", "transport"},
+		},
+		{
+			proxyType: "http",
+			cfg: &v1.HTTPProxyConfig{
+				ProxyBaseConfig:   newV2ProxyTestBaseConfig(t, "http"),
+				DomainConfig:      v1.DomainConfig{CustomDomains: []string{"app.example.com"}, SubDomain: "app"},
+				Locations:         []string{"/api"},
+				HTTPUser:          "secret-http-user",
+				HTTPPassword:      "secret-http-password",
+				HostHeaderRewrite: "backend.example.com",
+				RequestHeaders:    v1.HeaderOperations{Set: map[string]string{"X-Secret": "secret-request-header"}},
+				ResponseHeaders:   v1.HeaderOperations{Set: map[string]string{"X-Secret": "secret-response-header"}},
+				RouteByHTTPUser:   "secret-http-route-user",
+			},
+			blockKeys: []string{"annotations", "customDomains", "hostHeaderRewrite", "loadBalancer", "locations", "metadatas", "subdomain", "transport"},
+		},
+		{
+			proxyType: "https",
+			cfg: &v1.HTTPSProxyConfig{
+				ProxyBaseConfig: newV2ProxyTestBaseConfig(t, "https"),
+				DomainConfig:    v1.DomainConfig{CustomDomains: []string{"secure.example.com"}, SubDomain: "secure"},
+			},
+			blockKeys: []string{"annotations", "customDomains", "loadBalancer", "metadatas", "subdomain", "transport"},
+		},
+		{
+			proxyType: "tcpmux",
+			cfg: &v1.TCPMuxProxyConfig{
+				ProxyBaseConfig: newV2ProxyTestBaseConfig(t, "tcpmux"),
+				DomainConfig:    v1.DomainConfig{CustomDomains: []string{"mux.example.com"}, SubDomain: "mux"},
+				HTTPUser:        strings.Join([]string{"secret", "mux-http-user"}, "-"),
+				HTTPPassword:    strings.Join([]string{"secret", "mux-http-password"}, "-"),
+				RouteByHTTPUser: "displayed-mux-user",
+				Multiplexer:     "httpconnect",
+			},
+			blockKeys: []string{"annotations", "customDomains", "loadBalancer", "metadatas", "multiplexer", "routeByHTTPUser", "subdomain", "transport"},
+		},
+		{
+			proxyType: "stcp",
+			cfg: &v1.STCPProxyConfig{
+				ProxyBaseConfig: newV2ProxyTestBaseConfig(t, "stcp"),
+				Secretkey:       strings.Join([]string{"secret", "stcp-key"}, "-"),
+				AllowUsers:      []string{strings.Join([]string{"secret", "stcp-user"}, "-")},
+			},
+			blockKeys: []string{"annotations", "loadBalancer", "metadatas", "transport"},
+		},
+		{
+			proxyType: "sudp",
+			cfg: &v1.SUDPProxyConfig{
+				ProxyBaseConfig: newV2ProxyTestBaseConfig(t, "sudp"),
+				Secretkey:       strings.Join([]string{"secret", "sudp-key"}, "-"),
+				AllowUsers:      []string{strings.Join([]string{"secret", "sudp-user"}, "-")},
+			},
+			blockKeys: []string{"annotations", "loadBalancer", "metadatas", "transport"},
+		},
+		{
+			proxyType: "xtcp",
+			cfg: &v1.XTCPProxyConfig{
+				ProxyBaseConfig: newV2ProxyTestBaseConfig(t, "xtcp"),
+				Secretkey:       strings.Join([]string{"secret", "xtcp-key"}, "-"),
+				AllowUsers:      []string{strings.Join([]string{"secret", "xtcp-user"}, "-")},
+			},
+			blockKeys: []string{"annotations", "loadBalancer", "metadatas", "transport"},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.proxyType, func(t *testing.T) {
+			spec := buildV2ProxySpec(tt.proxyType, tt.cfg)
+			raw := mustMarshalJSON(t, spec)
+
+			var specObject map[string]json.RawMessage
+			if err := json.Unmarshal(raw, &specObject); err != nil {
+				t.Fatalf("unmarshal spec failed: %v", err)
+			}
+			assertRawJSONKeys(t, specObject, tt.proxyType, "type")
+
+			var gotType string
+			if err := json.Unmarshal(specObject["type"], &gotType); err != nil {
+				t.Fatalf("unmarshal spec type failed: %v", err)
+			}
+			if gotType != tt.proxyType {
+				t.Fatalf("spec type mismatch, want %q got %q", tt.proxyType, gotType)
+			}
+
+			var block map[string]json.RawMessage
+			if err := json.Unmarshal(specObject[tt.proxyType], &block); err != nil {
+				t.Fatalf("unmarshal active block failed: %v", err)
+			}
+			assertRawJSONKeys(t, block, tt.blockKeys...)
+			assertV2ProxyCommonSpec(t, block)
+			assertV2ProxyTypeFields(t, tt.proxyType, specObject[tt.proxyType])
+			assertNoV2ProxySensitiveFields(t, block)
+
+			content := string(raw)
+			for _, secret := range []string{
+				"secret-proxy-name",
+				"secret-group-key",
+				"secret-local-host",
+				"secret-plugin-user",
+				"secret-plugin-password",
+				"secret-health-path",
+				"secret-http-user",
+				"secret-http-password",
+				"secret-request-header",
+				"secret-response-header",
+				"secret-http-route-user",
+				"secret-mux-http-user",
+				"secret-mux-http-password",
+				"secret-stcp-key",
+				"secret-stcp-user",
+				"secret-sudp-key",
+				"secret-sudp-user",
+				"secret-xtcp-key",
+				"secret-xtcp-user",
+			} {
+				if strings.Contains(content, secret) {
+					t.Fatalf("sensitive value %q leaked in spec: %s", secret, content)
+				}
+			}
+		})
+	}
+}
+
+func assertV2ProxyTypeFields(t *testing.T, proxyType string, raw json.RawMessage) {
+	t.Helper()
+
+	switch proxyType {
+	case "tcp":
+		var block model.V2TCPProxySpec
+		if err := json.Unmarshal(raw, &block); err != nil {
+			t.Fatalf("unmarshal tcp block failed: %v", err)
+		}
+		if block.RemotePort == nil || *block.RemotePort != 6000 {
+			t.Fatalf("tcp remote port mismatch: %#v", block.RemotePort)
+		}
+	case "udp":
+		var block model.V2UDPProxySpec
+		if err := json.Unmarshal(raw, &block); err != nil {
+			t.Fatalf("unmarshal udp block failed: %v", err)
+		}
+		if block.RemotePort == nil || *block.RemotePort != 7000 {
+			t.Fatalf("udp remote port mismatch: %#v", block.RemotePort)
+		}
+	case "http":
+		var block model.V2HTTPProxySpec
+		if err := json.Unmarshal(raw, &block); err != nil {
+			t.Fatalf("unmarshal http block failed: %v", err)
+		}
+		if len(block.CustomDomains) != 1 || block.CustomDomains[0] != "app.example.com" ||
+			block.Subdomain != "app" || len(block.Locations) != 1 || block.Locations[0] != "/api" ||
+			block.HostHeaderRewrite != "backend.example.com" {
+			t.Fatalf("http fields mismatch: %#v", block)
+		}
+	case "https":
+		var block model.V2HTTPSProxySpec
+		if err := json.Unmarshal(raw, &block); err != nil {
+			t.Fatalf("unmarshal https block failed: %v", err)
+		}
+		if len(block.CustomDomains) != 1 || block.CustomDomains[0] != "secure.example.com" || block.Subdomain != "secure" {
+			t.Fatalf("https fields mismatch: %#v", block)
+		}
+	case "tcpmux":
+		var block model.V2TCPMuxProxySpec
+		if err := json.Unmarshal(raw, &block); err != nil {
+			t.Fatalf("unmarshal tcpmux block failed: %v", err)
+		}
+		if len(block.CustomDomains) != 1 || block.CustomDomains[0] != "mux.example.com" ||
+			block.Subdomain != "mux" || block.Multiplexer != "httpconnect" || block.RouteByHTTPUser != "displayed-mux-user" {
+			t.Fatalf("tcpmux fields mismatch: %#v", block)
+		}
+	}
+}
+
+func TestBuildV2ProxyRespOfflineTypedShells(t *testing.T) {
+	for _, proxyType := range apiV2ProxyTypes {
+		t.Run(proxyType, func(t *testing.T) {
+			resp := (&Controller{}).buildV2ProxyResp(&mem.ProxyStats{
+				Name: "offline-" + proxyType,
+				Type: proxyType,
+			})
+			if resp.Status.State != "offline" {
+				t.Fatalf("offline phase mismatch: %#v", resp.Status)
+			}
+
+			var specObject map[string]json.RawMessage
+			if err := json.Unmarshal(mustMarshalJSON(t, resp.Spec), &specObject); err != nil {
+				t.Fatalf("unmarshal offline spec failed: %v", err)
+			}
+			assertRawJSONKeys(t, specObject, proxyType, "type")
+			assertRawJSONKeysFromMessage(t, specObject[proxyType])
+		})
+	}
+}
+
+func TestBuildV2ProxySpecDoesNotPopulateMismatchedBlock(t *testing.T) {
+	spec := buildV2ProxySpec("tcp", &v1.UDPProxyConfig{
+		ProxyBaseConfig: newV2ProxyTestBaseConfig(t, "udp"),
+		RemotePort:      7000,
+	})
+
+	var specObject map[string]json.RawMessage
+	if err := json.Unmarshal(mustMarshalJSON(t, spec), &specObject); err != nil {
+		t.Fatalf("unmarshal mismatched spec failed: %v", err)
+	}
+	assertRawJSONKeys(t, specObject, "tcp", "type")
+	assertRawJSONKeysFromMessage(t, specObject["tcp"])
+}
+
+func newV2ProxyTestBaseConfig(t *testing.T, proxyType string) v1.ProxyBaseConfig {
+	t.Helper()
+
+	bandwidthLimit, err := configtypes.NewBandwidthQuantity("10MB")
+	if err != nil {
+		t.Fatalf("create bandwidth limit failed: %v", err)
+	}
+	enabled := false
+	return v1.ProxyBaseConfig{
+		Name:        "secret-proxy-name",
+		Type:        proxyType,
+		Enabled:     &enabled,
+		Annotations: map[string]string{"annotation-key": "annotation-value"},
+		Metadatas:   map[string]string{"metadata-key": "metadata-value"},
+		Transport: v1.ProxyTransport{
+			UseEncryption:        true,
+			UseCompression:       true,
+			BandwidthLimit:       bandwidthLimit,
+			BandwidthLimitMode:   configtypes.BandwidthLimitModeServer,
+			ProxyProtocolVersion: "v2",
+		},
+		LoadBalancer: v1.LoadBalancerConfig{
+			Group:    "public-group",
+			GroupKey: "secret-group-key",
+		},
+		HealthCheck: v1.HealthCheckConfig{
+			Type: "http",
+			Path: "secret-health-path",
+		},
+		ProxyBackend: v1.ProxyBackend{
+			LocalIP:   "secret-local-host",
+			LocalPort: 8080,
+			Plugin: v1.TypedClientPluginOptions{
+				Type: v1.PluginHTTPProxy,
+				ClientPluginOptions: &v1.HTTPProxyPluginOptions{
+					Type:         v1.PluginHTTPProxy,
+					HTTPUser:     "secret-plugin-user",
+					HTTPPassword: "secret-plugin-password",
+				},
+			},
+		},
+	}
+}
+
+func assertV2ProxyCommonSpec(t *testing.T, block map[string]json.RawMessage) {
+	t.Helper()
+
+	var annotations map[string]string
+	if err := json.Unmarshal(block["annotations"], &annotations); err != nil {
+		t.Fatalf("unmarshal annotations failed: %v", err)
+	}
+	if annotations["annotation-key"] != "annotation-value" {
+		t.Fatalf("annotations mismatch: %#v", annotations)
+	}
+
+	var metadatas map[string]string
+	if err := json.Unmarshal(block["metadatas"], &metadatas); err != nil {
+		t.Fatalf("unmarshal metadatas failed: %v", err)
+	}
+	if metadatas["metadata-key"] != "metadata-value" {
+		t.Fatalf("metadatas mismatch: %#v", metadatas)
+	}
+
+	assertRawJSONKeysFromMessage(t, block["transport"],
+		"bandwidthLimit",
+		"bandwidthLimitMode",
+		"useCompression",
+		"useEncryption",
+	)
+	var transport model.V2ProxyTransportSpec
+	if err := json.Unmarshal(block["transport"], &transport); err != nil {
+		t.Fatalf("unmarshal transport failed: %v", err)
+	}
+	if !transport.UseEncryption || !transport.UseCompression ||
+		transport.BandwidthLimit != "10MB" || transport.BandwidthLimitMode != "server" {
+		t.Fatalf("transport mismatch: %#v", transport)
+	}
+
+	assertRawJSONKeysFromMessage(t, block["loadBalancer"], "group")
+	var loadBalancer model.V2ProxyLoadBalancerSpec
+	if err := json.Unmarshal(block["loadBalancer"], &loadBalancer); err != nil {
+		t.Fatalf("unmarshal load balancer failed: %v", err)
+	}
+	if loadBalancer.Group != "public-group" {
+		t.Fatalf("load balancer mismatch: %#v", loadBalancer)
+	}
+}
+
+func assertNoV2ProxySensitiveFields(t *testing.T, value any) {
+	t.Helper()
+
+	forbidden := map[string]struct{}{
+		"allowUsers":           {},
+		"enabled":              {},
+		"groupKey":             {},
+		"healthCheck":          {},
+		"httpPassword":         {},
+		"httpUser":             {},
+		"localIP":              {},
+		"localPort":            {},
+		"name":                 {},
+		"natTraversal":         {},
+		"plugin":               {},
+		"proxyProtocolVersion": {},
+		"requestHeaders":       {},
+		"responseHeaders":      {},
+		"secretKey":            {},
+		"type":                 {},
+	}
+
+	var walk func(any)
+	walk = func(current any) {
+		switch current := current.(type) {
+		case map[string]any:
+			for key, nested := range current {
+				if _, ok := forbidden[key]; ok {
+					t.Fatalf("sensitive field %q leaked in active block", key)
+				}
+				walk(nested)
+			}
+		case []any:
+			for _, nested := range current {
+				walk(nested)
+			}
+		}
+	}
+
+	raw, err := json.Marshal(value)
+	if err != nil {
+		t.Fatalf("marshal active block failed: %v", err)
+	}
+	var decoded any
+	if err := json.Unmarshal(raw, &decoded); err != nil {
+		t.Fatalf("decode active block failed: %v", err)
+	}
+	walk(decoded)
+}

server/http/controller_v2_test.go

@@ -22,6 +22,7 @@ import (
 	"net/http/httptest"
 	"net/url"
 	"testing"
+	"time"
 
 	"github.com/gorilla/mux"
 
@@ -43,6 +44,7 @@ type v2EnvelopeForTest[T any] struct {
 type fakeStatsCollector struct {
 	server    *mem.ServerStats
 	proxies   map[string]*mem.ProxyStats
+	traffic   map[string]*mem.ProxyTrafficInfo
 	pruneable map[string]bool
 }
 
@@ -76,7 +78,7 @@ func (f *fakeStatsCollector) GetProxyByName(proxyName string) *mem.ProxyStats {
 }
 
 func (f *fakeStatsCollector) GetProxyTraffic(name string) *mem.ProxyTrafficInfo {
-	return nil
+	return f.traffic[name]
 }
 
 func (f *fakeStatsCollector) ClearOfflineProxies() (int, int) {
@@ -414,15 +416,42 @@ func TestAPIV2ProxyListDetailAndUsers(t *testing.T) {
 		t.Fatalf("proxy filter total mismatch: %#v", proxyResp.Data)
 	}
 	proxyItem := proxyResp.Data.Items[0]
-	if proxyItem.Name != "tcp-empty" || proxyItem.Type != "tcp" || proxyItem.User != "" || proxyItem.Status.State != "offline" {
+	if proxyItem.Name != "tcp-empty" || proxyItem.Spec.Type != "tcp" || proxyItem.User != "" || proxyItem.Status.State != "offline" {
 		t.Fatalf("proxy item mismatch: %#v", proxyItem)
 	}
+	rawProxyResp := decodeResponse[v2EnvelopeForTest[model.V2PageResp[map[string]json.RawMessage]]](t, resp)
+	assertRawJSONKeys(t, rawProxyResp.Data.Items[0], "clientID", "name", "spec", "status", "user")
+	var rawListSpec map[string]json.RawMessage
+	if err := json.Unmarshal(rawProxyResp.Data.Items[0]["spec"], &rawListSpec); err != nil {
+		t.Fatalf("unmarshal list proxy spec failed: %v", err)
+	}
+	assertRawJSONKeys(t, rawListSpec, "tcp", "type")
+	assertRawJSONKeysFromMessage(t, rawListSpec["tcp"])
 
 	resp = performRequest(router, "/api/v2/proxies/tcp-alice")
+	rawProxyDetailResp := decodeResponse[v2EnvelopeForTest[map[string]json.RawMessage]](t, resp)
+	assertRawJSONKeysFromMessage(t, rawProxyDetailResp.Data["status"],
+		"curConns",
+		"lastCloseAt",
+		"lastStartAt",
+		"phase",
+		"todayTrafficIn",
+		"todayTrafficOut",
+	)
 	proxyDetailResp := decodeResponse[v2EnvelopeForTest[model.V2ProxyResp]](t, resp)
 	if proxyDetailResp.Data.Name != "tcp-alice" || proxyDetailResp.Data.User != "alice" {
 		t.Fatalf("proxy detail mismatch: %#v", proxyDetailResp.Data)
 	}
+	assertRawJSONKeys(t, rawProxyDetailResp.Data, "clientID", "name", "spec", "status", "user")
+	var rawDetailSpec map[string]json.RawMessage
+	if err := json.Unmarshal(rawProxyDetailResp.Data["spec"], &rawDetailSpec); err != nil {
+		t.Fatalf("unmarshal detail proxy spec failed: %v", err)
+	}
+	assertRawJSONKeys(t, rawDetailSpec, "tcp", "type")
+	assertRawJSONKeysFromMessage(t, rawDetailSpec["tcp"])
+	if proxyDetailResp.Data.Status.LastStartAt != 1783504200 || proxyDetailResp.Data.Status.LastCloseAt != 1783504300 {
+		t.Fatalf("proxy detail timestamp mismatch: %#v", proxyDetailResp.Data.Status)
+	}
 
 	resp = performRequest(router, "/api/v2/users?page=1&pageSize=50")
 	userResp := decodeResponse[v2EnvelopeForTest[model.V2PageResp[model.V2UserResp]]](t, resp)
@@ -441,6 +470,140 @@ func TestAPIV2ProxyListDetailAndUsers(t *testing.T) {
 	}
 }
 
+func TestAPIV2ProxyTrafficEnvelopeSchemaAndHistory(t *testing.T) {
+	oldStatsCollector := mem.StatsCollector
+	mem.StatsCollector = &fakeStatsCollector{
+		proxies: map[string]*mem.ProxyStats{
+			"ssh": {Name: "ssh", Type: "tcp"},
+		},
+		traffic: map[string]*mem.ProxyTrafficInfo{
+			"ssh": {
+				Name:       "ssh",
+				TrafficIn:  []int64{70, 60, 50, 40, 30, 20, 10},
+				TrafficOut: []int64{700, 600, 500, 400, 300, 200, 100},
+			},
+		},
+	}
+	t.Cleanup(func() {
+		mem.StatsCollector = oldStatsCollector
+	})
+
+	controller := NewController(&v1.ServerConfig{}, registry.NewClientRegistry(), serverproxy.NewManager())
+	router := newV2TestRouter(controller)
+
+	resp := performRequest(router, "/api/v2/proxies/ssh/traffic")
+	if resp.Code != http.StatusOK {
+		t.Fatalf("status mismatch, want %d got %d, body: %s", http.StatusOK, resp.Code, resp.Body.String())
+	}
+	rawResp := decodeResponse[v2EnvelopeForTest[map[string]json.RawMessage]](t, resp)
+	if rawResp.Code != http.StatusOK || rawResp.Msg != "success" {
+		t.Fatalf("envelope mismatch: %#v", rawResp)
+	}
+	assertRawJSONKeys(t, rawResp.Data, "granularity", "history", "name", "unit")
+
+	trafficResp := decodeResponse[v2EnvelopeForTest[model.V2ProxyTrafficResp]](t, resp)
+	if trafficResp.Data.Name != "ssh" || trafficResp.Data.Unit != "bytes" || trafficResp.Data.Granularity != "day" {
+		t.Fatalf("traffic metadata mismatch: %#v", trafficResp.Data)
+	}
+	if len(trafficResp.Data.History) != 7 {
+		t.Fatalf("history length mismatch, want 7 got %d: %#v", len(trafficResp.Data.History), trafficResp.Data.History)
+	}
+
+	wantIn := []int64{10, 20, 30, 40, 50, 60, 70}
+	wantOut := []int64{100, 200, 300, 400, 500, 600, 700}
+	var prevDate time.Time
+	for i, point := range trafficResp.Data.History {
+		assertRawJSONKeysFromMessage(t, mustMarshalJSON(t, point), "date", "trafficIn", "trafficOut")
+		if point.TrafficIn != wantIn[i] || point.TrafficOut != wantOut[i] {
+			t.Fatalf("history[%d] traffic mismatch: %#v", i, point)
+		}
+		parsedDate, err := time.Parse(time.DateOnly, point.Date)
+		if err != nil {
+			t.Fatalf("history[%d] date should be yyyy-mm-dd, got %q: %v", i, point.Date, err)
+		}
+		if i > 0 && !parsedDate.Equal(prevDate.AddDate(0, 0, 1)) {
+			t.Fatalf("history dates should be oldest to newest, got %s after %s", point.Date, prevDate.Format(time.DateOnly))
+		}
+		prevDate = parsedDate
+	}
+}
+
+func TestAPIV2ProxyTrafficNotFoundEnvelope(t *testing.T) {
+	controller := newV2TestController(t)
+	router := newV2TestRouter(controller)
+
+	resp := performRequest(router, "/api/v2/proxies/missing/traffic")
+	if resp.Code != http.StatusNotFound {
+		t.Fatalf("status mismatch, want %d got %d, body: %s", http.StatusNotFound, resp.Code, resp.Body.String())
+	}
+	errResp := decodeResponse[httppkg.V2Response](t, resp)
+	if errResp.Code != http.StatusNotFound || errResp.Msg != "no proxy info found" || errResp.Data != nil {
+		t.Fatalf("not found envelope mismatch: %#v", errResp)
+	}
+}
+
+func TestAPIV2ProxyDetailAndTrafficEncodedName(t *testing.T) {
+	name := "folder/ssh?x#y"
+	oldStatsCollector := mem.StatsCollector
+	mem.StatsCollector = &fakeStatsCollector{
+		proxies: map[string]*mem.ProxyStats{
+			name: {Name: name, Type: "tcp", User: "encoded"},
+		},
+		traffic: map[string]*mem.ProxyTrafficInfo{
+			name: {
+				Name:       name,
+				TrafficIn:  []int64{1},
+				TrafficOut: []int64{2},
+			},
+		},
+	}
+	t.Cleanup(func() {
+		mem.StatsCollector = oldStatsCollector
+	})
+
+	controller := NewController(&v1.ServerConfig{}, registry.NewClientRegistry(), serverproxy.NewManager())
+	router := newV2TestRouter(controller)
+	encodedName := url.PathEscape(name)
+
+	resp := performRequest(router, "/api/v2/proxies/"+encodedName)
+	if resp.Code != http.StatusOK {
+		t.Fatalf("encoded proxy detail status mismatch, want %d got %d, body: %s", http.StatusOK, resp.Code, resp.Body.String())
+	}
+	detailResp := decodeResponse[v2EnvelopeForTest[model.V2ProxyResp]](t, resp)
+	if detailResp.Data.Name != name || detailResp.Data.User != "encoded" {
+		t.Fatalf("encoded proxy detail mismatch: %#v", detailResp.Data)
+	}
+
+	resp = performRequest(router, "/api/v2/proxies/"+encodedName+"/traffic")
+	if resp.Code != http.StatusOK {
+		t.Fatalf("encoded traffic status mismatch, want %d got %d, body: %s", http.StatusOK, resp.Code, resp.Body.String())
+	}
+	trafficResp := decodeResponse[v2EnvelopeForTest[model.V2ProxyTrafficResp]](t, resp)
+	if trafficResp.Data.Name != name {
+		t.Fatalf("encoded traffic name mismatch: %#v", trafficResp.Data)
+	}
+	if got := trafficResp.Data.History[len(trafficResp.Data.History)-1]; got.TrafficIn != 1 || got.TrafficOut != 2 {
+		t.Fatalf("encoded traffic latest point mismatch: %#v", got)
+	}
+}
+
+func TestAPIV2ProxyTrafficInvalidEncodedNameUses400Envelope(t *testing.T) {
+	controller := newV2TestController(t)
+	handler := httppkg.MakeHTTPHandlerFuncV2(controller.APIV2ProxyTraffic)
+	req := httptest.NewRequest(http.MethodGet, "/api/v2/proxies/%25ZZ/traffic", nil)
+	req = mux.SetURLVars(req, map[string]string{"name": "%ZZ"})
+	resp := httptest.NewRecorder()
+	handler.ServeHTTP(resp, req)
+
+	if resp.Code != http.StatusBadRequest {
+		t.Fatalf("status mismatch, want %d got %d, body: %s", http.StatusBadRequest, resp.Code, resp.Body.String())
+	}
+	errResp := decodeResponse[httppkg.V2Response](t, resp)
+	if errResp.Code != http.StatusBadRequest || errResp.Msg != "invalid proxy name" || errResp.Data != nil {
+		t.Fatalf("invalid encoded name envelope mismatch: %#v", errResp)
+	}
+}
+
 func TestMatchV2ProxyQueryMatchesSpecFields(t *testing.T) {
 	tests := []struct {
 		name string
@@ -450,66 +613,85 @@ func TestMatchV2ProxyQueryMatchesSpecFields(t *testing.T) {
 	}{
 		{
 			name: "tcp remote port",
-			item: model.V2ProxyResp{Name: "tcp-proxy", Type: "tcp", Spec: &model.TCPOutConf{
-				RemotePort: 6000,
+			item: model.V2ProxyResp{Name: "tcp-proxy", Spec: model.V2ProxySpec{
+				Type: "tcp",
+				TCP:  &model.V2TCPProxySpec{RemotePort: v2TestIntPtr(6000)},
 			}},
 			q:    "6000",
 			want: true,
 		},
 		{
 			name: "udp remote port",
-			item: model.V2ProxyResp{Name: "udp-proxy", Type: "udp", Spec: &model.UDPOutConf{
-				RemotePort: 7000,
+			item: model.V2ProxyResp{Name: "udp-proxy", Spec: model.V2ProxySpec{
+				Type: "udp",
+				UDP:  &model.V2UDPProxySpec{RemotePort: v2TestIntPtr(7000)},
 			}},
 			q:    "7000",
 			want: true,
 		},
 		{
 			name: "remote port does not match colon form",
-			item: model.V2ProxyResp{Name: "tcp-proxy", Type: "tcp", Spec: &model.TCPOutConf{
-				RemotePort: 6000,
+			item: model.V2ProxyResp{Name: "tcp-proxy", Spec: model.V2ProxySpec{
+				Type: "tcp",
+				TCP:  &model.V2TCPProxySpec{RemotePort: v2TestIntPtr(6000)},
 			}},
 			q:    ":6000",
 			want: false,
 		},
 		{
 			name: "http custom domain",
-			item: model.V2ProxyResp{Name: "http-proxy", Type: "http", Spec: &model.HTTPOutConf{
-				DomainConfig: v1.DomainConfig{CustomDomains: []string{"app.example.com"}},
+			item: model.V2ProxyResp{Name: "http-proxy", Spec: model.V2ProxySpec{
+				Type: "http",
+				HTTP: &model.V2HTTPProxySpec{CustomDomains: []string{"app.example.com"}},
 			}},
 			q:    "app.example.com",
 			want: true,
 		},
 		{
 			name: "https subdomain",
-			item: model.V2ProxyResp{Name: "https-proxy", Type: "https", Spec: &model.HTTPSOutConf{
-				DomainConfig: v1.DomainConfig{SubDomain: "portal"},
+			item: model.V2ProxyResp{Name: "https-proxy", Spec: model.V2ProxySpec{
+				Type:  "https",
+				HTTPS: &model.V2HTTPSProxySpec{Subdomain: "portal"},
 			}},
 			q:    "portal",
 			want: true,
 		},
 		{
 			name: "subdomain does not match expanded host",
-			item: model.V2ProxyResp{Name: "https-proxy", Type: "https", Spec: &model.HTTPSOutConf{
-				DomainConfig: v1.DomainConfig{SubDomain: "portal"},
+			item: model.V2ProxyResp{Name: "https-proxy", Spec: model.V2ProxySpec{
+				Type:  "https",
+				HTTPS: &model.V2HTTPSProxySpec{Subdomain: "portal"},
 			}},
 			q:    "portal.example.com",
 			want: false,
 		},
 		{
 			name: "tcpmux custom domain",
-			item: model.V2ProxyResp{Name: "tcpmux-proxy", Type: "tcpmux", Spec: &model.TCPMuxOutConf{
-				DomainConfig: v1.DomainConfig{CustomDomains: []string{"mux.example.com"}},
+			item: model.V2ProxyResp{Name: "tcpmux-proxy", Spec: model.V2ProxySpec{
+				Type:   "tcpmux",
+				TCPMux: &model.V2TCPMuxProxySpec{CustomDomains: []string{"mux.example.com"}},
 			}},
 			q:    "mux.example.com",
 			want: true,
 		},
 		{
-			name: "nil spec does not match spec fields",
-			item: model.V2ProxyResp{Name: "offline-proxy", Type: "tcp", Spec: nil},
+			name: "offline shell does not match online spec fields",
+			item: model.V2ProxyResp{Name: "offline-proxy", Spec: model.V2ProxySpec{
+				Type: "tcp",
+				TCP:  &model.V2TCPProxySpec{},
+			}},
 			q:    "6000",
 			want: false,
 		},
+		{
+			name: "offline shell does not contribute zero remote port",
+			item: model.V2ProxyResp{Name: "offline-proxy", Spec: model.V2ProxySpec{
+				Type: "tcp",
+				TCP:  &model.V2TCPProxySpec{},
+			}},
+			q:    "0",
+			want: false,
+		},
 	}
 
 	for _, tt := range tests {
@@ -565,6 +747,28 @@ func TestLegacyAPIResponsesRemainBare(t *testing.T) {
 	if err := json.Unmarshal(resp.Body.Bytes(), &envelope); err == nil && envelope.Code != 0 {
 		t.Fatalf("legacy proxy response should not use v2 envelope: %#v", envelope)
 	}
+
+	resp = performRequest(router, "/api/traffic/tcp-alice")
+	var traffic model.GetProxyTrafficResp
+	if err := json.Unmarshal(resp.Body.Bytes(), &traffic); err != nil {
+		t.Fatalf("legacy traffic should be a bare object: %v, body: %s", err, resp.Body.String())
+	}
+	if traffic.Name != "tcp-alice" ||
+		len(traffic.TrafficIn) != 2 || traffic.TrafficIn[0] != 7 || traffic.TrafficIn[1] != 6 ||
+		len(traffic.TrafficOut) != 2 || traffic.TrafficOut[0] != 70 || traffic.TrafficOut[1] != 60 {
+		t.Fatalf("legacy traffic should preserve today-first arrays, got: %#v", traffic)
+	}
+	var trafficRaw map[string]json.RawMessage
+	if err := json.Unmarshal(resp.Body.Bytes(), &trafficRaw); err != nil {
+		t.Fatalf("unmarshal legacy traffic object failed: %v", err)
+	}
+	if _, ok := trafficRaw["data"]; ok {
+		t.Fatalf("legacy traffic should not use v2 envelope: %s", resp.Body.String())
+	}
+}
+
+func v2TestIntPtr(value int) *int {
+	return &value
 }
 
 func newV2TestController(t *testing.T) *Controller {
@@ -590,6 +794,10 @@ func newV2TestController(t *testing.T) *Controller {
 				TodayTrafficIn:  30,
 				TodayTrafficOut: 40,
 				CurConns:        2,
+				LastStartTime:   "07-08 12:30:00",
+				LastCloseTime:   "07-08 12:31:40",
+				LastStartAt:     1783504200,
+				LastCloseAt:     1783504300,
 			},
 			"http-alice": {
 				Name:     "http-alice",
@@ -605,6 +813,13 @@ func newV2TestController(t *testing.T) *Controller {
 				ClientID: "client-b",
 			},
 		},
+		traffic: map[string]*mem.ProxyTrafficInfo{
+			"tcp-alice": {
+				Name:       "tcp-alice",
+				TrafficIn:  []int64{7, 6},
+				TrafficOut: []int64{70, 60},
+			},
+		},
 	}
 	t.Cleanup(func() {
 		mem.StatsCollector = oldStatsCollector
@@ -629,10 +844,12 @@ func newV2TestRouter(controller *Controller) *mux.Router {
 	encodedPathRouter.UseEncodedPath()
 	encodedPathRouter.HandleFunc("/api/v2/clients/{key}", httppkg.MakeHTTPHandlerFuncV2(controller.APIV2ClientDetail)).Methods(http.MethodGet)
 	router.HandleFunc("/api/v2/proxies", httppkg.MakeHTTPHandlerFuncV2(controller.APIV2ProxyList)).Methods(http.MethodGet)
-	router.HandleFunc("/api/v2/proxies/{name}", httppkg.MakeHTTPHandlerFuncV2(controller.APIV2ProxyDetail)).Methods(http.MethodGet)
+	encodedPathRouter.HandleFunc("/api/v2/proxies/{name}/traffic", httppkg.MakeHTTPHandlerFuncV2(controller.APIV2ProxyTraffic)).Methods(http.MethodGet)
+	encodedPathRouter.HandleFunc("/api/v2/proxies/{name}", httppkg.MakeHTTPHandlerFuncV2(controller.APIV2ProxyDetail)).Methods(http.MethodGet)
 	router.HandleFunc("/api/serverinfo", httppkg.MakeHTTPHandlerFunc(controller.APIServerInfo)).Methods(http.MethodGet)
 	router.HandleFunc("/api/clients", httppkg.MakeHTTPHandlerFunc(controller.APIClientList)).Methods(http.MethodGet)
 	router.HandleFunc("/api/proxy/{type}", httppkg.MakeHTTPHandlerFunc(controller.APIProxyByType)).Methods(http.MethodGet)
+	router.HandleFunc("/api/traffic/{name}", httppkg.MakeHTTPHandlerFunc(controller.APIProxyTraffic)).Methods(http.MethodGet)
 	return router
 }
 
@@ -679,3 +896,13 @@ func assertRawJSONKeysFromMessage(t *testing.T, raw json.RawMessage, want ...str
 	}
 	assertRawJSONKeys(t, out, want...)
 }
+
+func mustMarshalJSON(t *testing.T, value any) json.RawMessage {
+	t.Helper()
+
+	out, err := json.Marshal(value)
+	if err != nil {
+		t.Fatalf("marshal json failed: %v", err)
+	}
+	return out
+}

server/http/model/v2.go

@@ -75,18 +75,105 @@ type V2ClientStatusResp struct {
 
 type V2ProxyResp struct {
 	Name     string            `json:"name"`
-	Type     string            `json:"type"`
 	User     string            `json:"user"`
 	ClientID string            `json:"clientID"`
-	Spec     any               `json:"spec"`
+	Spec     V2ProxySpec       `json:"spec"`
 	Status   V2ProxyStatusResp `json:"status"`
 }
 
+type V2ProxySpec struct {
+	Type string `json:"type"`
+
+	TCP    *V2TCPProxySpec    `json:"tcp,omitempty"`
+	UDP    *V2UDPProxySpec    `json:"udp,omitempty"`
+	HTTP   *V2HTTPProxySpec   `json:"http,omitempty"`
+	HTTPS  *V2HTTPSProxySpec  `json:"https,omitempty"`
+	TCPMux *V2TCPMuxProxySpec `json:"tcpmux,omitempty"`
+	STCP   *V2STCPProxySpec   `json:"stcp,omitempty"`
+	SUDP   *V2SUDPProxySpec   `json:"sudp,omitempty"`
+	XTCP   *V2XTCPProxySpec   `json:"xtcp,omitempty"`
+}
+
+type V2ProxyBaseSpec struct {
+	Annotations  map[string]string        `json:"annotations,omitempty"`
+	Metadatas    map[string]string        `json:"metadatas,omitempty"`
+	Transport    *V2ProxyTransportSpec    `json:"transport,omitempty"`
+	LoadBalancer *V2ProxyLoadBalancerSpec `json:"loadBalancer,omitempty"`
+}
+
+type V2ProxyTransportSpec struct {
+	UseEncryption      bool   `json:"useEncryption"`
+	UseCompression     bool   `json:"useCompression"`
+	BandwidthLimit     string `json:"bandwidthLimit"`
+	BandwidthLimitMode string `json:"bandwidthLimitMode"`
+}
+
+type V2ProxyLoadBalancerSpec struct {
+	Group string `json:"group"`
+}
+
+type V2TCPProxySpec struct {
+	V2ProxyBaseSpec
+	RemotePort *int `json:"remotePort,omitempty"`
+}
+
+type V2UDPProxySpec struct {
+	V2ProxyBaseSpec
+	RemotePort *int `json:"remotePort,omitempty"`
+}
+
+type V2HTTPProxySpec struct {
+	V2ProxyBaseSpec
+	CustomDomains     []string `json:"customDomains,omitempty"`
+	Subdomain         string   `json:"subdomain,omitempty"`
+	Locations         []string `json:"locations,omitempty"`
+	HostHeaderRewrite string   `json:"hostHeaderRewrite,omitempty"`
+}
+
+type V2HTTPSProxySpec struct {
+	V2ProxyBaseSpec
+	CustomDomains []string `json:"customDomains,omitempty"`
+	Subdomain     string   `json:"subdomain,omitempty"`
+}
+
+type V2TCPMuxProxySpec struct {
+	V2ProxyBaseSpec
+	CustomDomains   []string `json:"customDomains,omitempty"`
+	Subdomain       string   `json:"subdomain,omitempty"`
+	Multiplexer     string   `json:"multiplexer,omitempty"`
+	RouteByHTTPUser string   `json:"routeByHTTPUser,omitempty"`
+}
+
+type V2STCPProxySpec struct {
+	V2ProxyBaseSpec
+}
+
+type V2SUDPProxySpec struct {
+	V2ProxyBaseSpec
+}
+
+type V2XTCPProxySpec struct {
+	V2ProxyBaseSpec
+}
+
 type V2ProxyStatusResp struct {
 	State           string `json:"phase"`
 	TodayTrafficIn  int64  `json:"todayTrafficIn"`
 	TodayTrafficOut int64  `json:"todayTrafficOut"`
 	CurConns        int64  `json:"curConns"`
-	LastStartTime   string `json:"lastStartTime"`
-	LastCloseTime   string `json:"lastCloseTime"`
+	LastStartAt     int64  `json:"lastStartAt,omitempty"`
+	LastCloseAt     int64  `json:"lastCloseAt,omitempty"`
+}
+
+type V2ProxyTrafficResp struct {
+	Name        string                    `json:"name"`
+	Unit        string                    `json:"unit"`
+	Granularity string                    `json:"granularity"`
+	History     []V2ProxyTrafficPointResp `json:"history"`
+}
+
+type V2ProxyTrafficPointResp struct {
+	Date       string `json:"date"`
+	TrafficIn  int64  `json:"trafficIn"`
+	TrafficOut int64  `json:"trafficOut"`
 }

server/registry/registry.go

@@ -28,6 +28,7 @@ type ClientInfo struct {
 	User             string
 	RawClientID      string
 	RunID            string
+	ControlID        uint64
 	Hostname         string
 	IP               string
 	Version          string
@@ -64,6 +65,16 @@ func newClientRegistryWithClock(clk clock.PassiveClock) *ClientRegistry {
 
 // Register stores/updates metadata for a client and returns the registry key plus whether it conflicts with an online client.
 func (cr *ClientRegistry) Register(user, rawClientID, runID, hostname, version, remoteAddr, wireProtocol string) (key string, conflict bool) {
+	return cr.RegisterWithControlID(user, rawClientID, runID, hostname, version, remoteAddr, wireProtocol, 0)
+}
+
+// RegisterWithControlID is the generation-aware form used by ControlManager.
+// A control ID is process-local and prevents an older control generation from
+// changing the registry entry now owned by a newer generation with the same run ID.
+func (cr *ClientRegistry) RegisterWithControlID(
+	user, rawClientID, runID, hostname, version, remoteAddr, wireProtocol string,
+	controlID uint64,
+) (key string, conflict bool) {
 	if runID == "" {
 		return "", false
 	}
@@ -83,6 +94,16 @@ func (cr *ClientRegistry) Register(user, rawClientID, runID, hostname, version,
 	if enforceUnique && exists && info.Online && info.RunID != "" && info.RunID != runID {
 		return key, true
 	}
+	if previousKey, ok := cr.runIndex[runID]; ok && previousKey != key {
+		if previous, ok := cr.clients[previousKey]; ok && previous.RunID == runID {
+			if previous.RawClientID == "" {
+				delete(cr.clients, previousKey)
+			} else {
+				setClientOffline(previous, now)
+			}
+		}
+		delete(cr.runIndex, runID)
+	}
 
 	if !exists {
 		info = &ClientInfo{
@@ -97,6 +118,7 @@ func (cr *ClientRegistry) Register(user, rawClientID, runID, hostname, version,
 
 	info.RawClientID = rawClientID
 	info.RunID = runID
+	info.ControlID = controlID
 	info.Hostname = hostname
 	info.IP = remoteAddr
 	info.Version = version
@@ -114,6 +136,16 @@ func (cr *ClientRegistry) Register(user, rawClientID, runID, hostname, version,
 
 // MarkOfflineByRunID marks the client as offline when the corresponding control disconnects.
 func (cr *ClientRegistry) MarkOfflineByRunID(runID string) {
+	cr.markOfflineByRunID(runID, 0, false)
+}
+
+// MarkOfflineByRunIDAndControlID marks a client offline only when the registry
+// entry still belongs to the supplied control generation.
+func (cr *ClientRegistry) MarkOfflineByRunIDAndControlID(runID string, controlID uint64) {
+	cr.markOfflineByRunID(runID, controlID, true)
+}
+
+func (cr *ClientRegistry) markOfflineByRunID(runID string, controlID uint64, matchControlID bool) {
 	cr.mu.Lock()
 	defer cr.mu.Unlock()
 
@@ -121,17 +153,23 @@ func (cr *ClientRegistry) MarkOfflineByRunID(runID string) {
 	if !ok {
 		return
 	}
-	if info, ok := cr.clients[key]; ok && info.RunID == runID {
+	if info, ok := cr.clients[key]; ok && info.RunID == runID && (!matchControlID || info.ControlID == controlID) {
 		if info.RawClientID == "" {
 			delete(cr.clients, key)
 		} else {
-			info.RunID = ""
-			info.Online = false
-			now := cr.clock.Now()
-			info.DisconnectedAt = now
+			setClientOffline(info, cr.clock.Now())
 		}
 	}
-	delete(cr.runIndex, runID)
+	if info, ok := cr.clients[key]; !ok || info.RunID != runID {
+		delete(cr.runIndex, runID)
+	}
+}
+
+func setClientOffline(info *ClientInfo, now time.Time) {
+	info.RunID = ""
+	info.ControlID = 0
+	info.Online = false
+	info.DisconnectedAt = now
 }
 
 // List returns a snapshot of all known clients.

server/registry/registry_test.go

@@ -72,3 +72,89 @@ func TestClientRegistryUsesClockForTimestamps(t *testing.T) {
 		t.Fatalf("disconnected time mismatch, want %s got %s", disconnectedAt, info.DisconnectedAt)
 	}
 }
+
+func TestClientRegistryControlIDPreventsStaleOffline(t *testing.T) {
+	registry := NewClientRegistry()
+	key, conflict := registry.RegisterWithControlID(
+		"user", "client-id", "run-id", "old-host", "1.0.0", "127.0.0.1", wire.ProtocolV1, 1,
+	)
+	if conflict {
+		t.Fatal("unexpected client conflict")
+	}
+	_, conflict = registry.RegisterWithControlID(
+		"user", "client-id", "run-id", "new-host", "1.0.1", "127.0.0.2", wire.ProtocolV2, 2,
+	)
+	if conflict {
+		t.Fatal("same run ID replacement should not conflict")
+	}
+
+	registry.MarkOfflineByRunIDAndControlID("run-id", 1)
+	info, ok := registry.GetByKey(key)
+	if !ok {
+		t.Fatalf("client %q not found", key)
+	}
+	if !info.Online || info.ControlID != 2 || info.Hostname != "new-host" {
+		t.Fatalf("stale offline changed current generation: %+v", info)
+	}
+
+	registry.MarkOfflineByRunIDAndControlID("run-id", 2)
+	info, ok = registry.GetByKey(key)
+	if !ok {
+		t.Fatalf("client %q not found after disconnect", key)
+	}
+	if info.Online || info.ControlID != 0 || info.RunID != "" {
+		t.Fatalf("current generation was not marked offline: %+v", info)
+	}
+}
+
+func TestClientRegistryClientIDConflictSemantics(t *testing.T) {
+	registry := NewClientRegistry()
+	_, conflict := registry.RegisterWithControlID(
+		"user", "client-id", "run-one", "host", "1.0.0", "127.0.0.1", wire.ProtocolV1, 1,
+	)
+	if conflict {
+		t.Fatal("unexpected initial client conflict")
+	}
+	_, conflict = registry.RegisterWithControlID(
+		"user", "client-id", "run-two", "host", "1.0.0", "127.0.0.2", wire.ProtocolV1, 2,
+	)
+	if !conflict {
+		t.Fatal("different online run IDs with the same explicit client ID must conflict")
+	}
+
+	registry.MarkOfflineByRunIDAndControlID("run-one", 1)
+	_, conflict = registry.RegisterWithControlID(
+		"user", "client-id", "run-two", "host", "1.0.0", "127.0.0.2", wire.ProtocolV1, 2,
+	)
+	if conflict {
+		t.Fatal("offline explicit client ID should be reusable")
+	}
+}
+
+func TestClientRegistrySameRunIDMovesBetweenClientKeys(t *testing.T) {
+	registry := NewClientRegistry()
+	oldKey, conflict := registry.RegisterWithControlID(
+		"user", "old-client", "run-id", "old-host", "1.0.0", "127.0.0.1", wire.ProtocolV1, 1,
+	)
+	if conflict {
+		t.Fatal("unexpected initial client conflict")
+	}
+	newKey, conflict := registry.RegisterWithControlID(
+		"user", "new-client", "run-id", "new-host", "1.0.1", "127.0.0.2", wire.ProtocolV2, 2,
+	)
+	if conflict {
+		t.Fatal("same run ID moving to a new client key should not conflict")
+	}
+
+	oldInfo, ok := registry.GetByKey(oldKey)
+	if !ok {
+		t.Fatalf("old explicit client %q should remain as offline history", oldKey)
+	}
+	if oldInfo.Online || oldInfo.RunID != "" || oldInfo.ControlID != 0 {
+		t.Fatalf("old client key remained online: %+v", oldInfo)
+	}
+	newInfo, ok := registry.GetByKey(newKey)
+	if !ok || !newInfo.Online || newInfo.RunID != "run-id" || newInfo.ControlID != 2 {
+		t.Fatalf("new client key was not registered: %+v", newInfo)
+	}
+}

server/service.go

@@ -18,6 +18,7 @@ import (
 	"bytes"
 	"context"
 	"crypto/tls"
+	"errors"
 	"fmt"
 	"io"
 	"net"
@@ -51,7 +52,6 @@ import (
 	"github.com/fatedier/frp/pkg/util/xlog"
 	"github.com/fatedier/frp/server/controller"
 	"github.com/fatedier/frp/server/group"
-	"github.com/fatedier/frp/server/metrics"
 	"github.com/fatedier/frp/server/ports"
 	"github.com/fatedier/frp/server/proxy"
 	"github.com/fatedier/frp/server/registry"
@@ -64,6 +64,8 @@ const (
 	vhostReadWriteTimeout time.Duration = 30 * time.Second
 )
 
+var errControlReplaced = errors.New("control was replaced during login")
+
 func init() {
 	crypto.DefaultSalt = "frp"
 	// Disable quic-go's receive buffer warning.
@@ -161,9 +163,10 @@ func NewService(cfg *v1.ServerConfig) (*Service, error) {
 		return nil, err
 	}
 
+	clientRegistry := registry.NewClientRegistry()
 	svr := &Service{
-		ctlManager:     NewControlManager(),
-		clientRegistry: registry.NewClientRegistry(),
+		ctlManager:     NewControlManager(clientRegistry),
+		clientRegistry: clientRegistry,
 		pxyManager:     proxy.NewManager(),
 		pluginManager:  plugin.NewManager(),
 		rc: &controller.ResourceController{
@@ -469,6 +472,9 @@ func (svr *Service) handleConnection(ctx context.Context, conn net.Conn, interna
 
 		if err != nil {
 			xl.Warnf("register control error: %v", err)
+			if ctl != nil {
+				svr.ctlManager.Remove(ctl)
+			}
 			if writeErr := writeWithDeadline(conn, connWriteTimeout, func() error {
 				return acceptedConn.conn.WriteMsg(&msg.LoginResp{
 					Version: version.Full(),
@@ -477,29 +483,27 @@ func (svr *Service) handleConnection(ctx context.Context, conn net.Conn, interna
 			}); writeErr != nil {
 				xl.Warnf("write login error response error: %v", writeErr)
 			}
-			conn.Close()
+			if ctl != nil {
+				_ = ctl.Close()
+			} else {
+				conn.Close()
+			}
 			return
 		}
-		if err = writeWithDeadline(conn, connWriteTimeout, func() error {
-			return acceptedConn.conn.WriteMsg(&msg.LoginResp{
-				Version: version.Full(),
-				RunID:   ctl.runID,
-				Error:   "",
+		if err = svr.completeControlLogin(ctl, func() error {
+			return writeWithDeadline(conn, connWriteTimeout, func() error {
+				return acceptedConn.conn.WriteMsg(&msg.LoginResp{
+					Version: version.Full(),
+					RunID:   ctl.runID,
+					Error:   "",
+				})
 			})
 		}); err != nil {
-			xl.Warnf("write login response error: %v", err)
-			svr.ctlManager.Del(m.RunID, ctl)
-			svr.clientRegistry.MarkOfflineByRunID(m.RunID)
-			conn.Close()
+			xl.Warnf("complete control login error: %v", err)
+			svr.ctlManager.Remove(ctl)
+			_ = ctl.Close()
 			return
 		}
-		ctl.Start()
-		metrics.Server.NewClient()
-		go func() {
-			// block until control closed
-			ctl.WaitClosed()
-			svr.ctlManager.Del(m.RunID, ctl)
-		}()
 	case *msg.NewWorkConn:
 		if err := svr.RegisterWorkConn(acceptedConn.conn, m); err != nil {
 			_ = acceptedConn.conn.WriteMsg(&msg.StartWorkConn{
@@ -527,6 +531,17 @@ func (svr *Service) handleConnection(ctx context.Context, conn net.Conn, interna
 	}
 }
 
+func (svr *Service) completeControlLogin(ctl *Control, writeSuccess func() error) error {
+	committed, err := svr.ctlManager.completeLogin(ctl, writeSuccess)
+	if err != nil {
+		return err
+	}
+	if !committed {
+		return errControlReplaced
+	}
+	return nil
+}
+
 type acceptedConnection struct {
 	conn          *msg.Conn
 	wireProtocol  string
@@ -768,16 +783,15 @@ func (svr *Service) RegisterControl(
 	}
 
 	ctl, err := NewControl(ctx, &SessionContext{
-		RC:             svr.rc,
-		PxyManager:     svr.pxyManager,
-		PluginManager:  svr.pluginManager,
-		AuthVerifier:   authVerifier,
-		EncryptionKey:  svr.auth.EncryptionKey(),
-		Conn:           ctlConn,
-		LoginMsg:       loginMsg,
-		ServerCfg:      svr.cfg,
-		ClientRegistry: svr.clientRegistry,
-		WireProtocol:   wireProtocol,
+		RC:            svr.rc,
+		PxyManager:    svr.pxyManager,
+		PluginManager: svr.pluginManager,
+		AuthVerifier:  authVerifier,
+		EncryptionKey: svr.auth.EncryptionKey(),
+		Conn:          ctlConn,
+		LoginMsg:      loginMsg,
+		ServerCfg:     svr.cfg,
+		WireProtocol:  wireProtocol,
 	})
 	if err != nil {
 		xl.Warnf("create new controller error: %v", err)
@@ -785,18 +799,17 @@ func (svr *Service) RegisterControl(
 		return nil, fmt.Errorf("unexpected error when creating new controller")
 	}
 
-	if oldCtl := svr.ctlManager.Add(loginMsg.RunID, ctl); oldCtl != nil {
-		oldCtl.WaitClosed()
+	if err := svr.ctlManager.Add(ctl); err != nil {
+		return ctl, err
 	}
+	ctl.WaitForHandoff()
 
-	remoteAddr := ctlConn.RemoteAddr().String()
-	if host, _, err := net.SplitHostPort(remoteAddr); err == nil {
-		remoteAddr = host
+	active, err := svr.ctlManager.Activate(ctl)
+	if err != nil {
+		return ctl, err
 	}
-	_, conflict := svr.clientRegistry.Register(loginMsg.User, loginMsg.ClientID, loginMsg.RunID, loginMsg.Hostname, loginMsg.Version, remoteAddr, wireProtocol)
-	if conflict {
-		svr.ctlManager.Del(loginMsg.RunID, ctl)
-		return nil, fmt.Errorf("client_id [%s] for user [%s] is already online", loginMsg.ClientID, loginMsg.User)
+	if !active {
+		return ctl, errControlReplaced
 	}
 
 	return ctl, nil
@@ -830,20 +843,25 @@ func (svr *Service) RegisterWorkConn(workConn *msg.Conn, newMsg *msg.NewWorkConn
 		xl.Warnf("invalid NewWorkConn with run id [%s]", newMsg.RunID)
 		return err
 	}
-	return ctl.RegisterWorkConn(proxy.NewWorkConn(workConn))
+	return svr.ctlManager.RegisterWorkConn(ctl, proxy.NewWorkConn(workConn))
 }
 
 func (svr *Service) RegisterVisitorConn(visitorConn net.Conn, newMsg *msg.NewVisitorConn, wireProtocol string) error {
-	visitorUser := ""
+	admit := func(visitorUser string) error {
+		return svr.rc.VisitorManager.NewConn(newMsg.ProxyName, visitorConn, newMsg.Timestamp, newMsg.SignKey,
+			newMsg.UseEncryption, newMsg.UseCompression, visitorUser, wireProtocol)
+	}
 	// TODO(deprecation): Compatible with old versions, can be without runID, user is empty. In later versions, it will be mandatory to include runID.
 	// If runID is required, it is not compatible with versions prior to v0.50.0.
 	if newMsg.RunID != "" {
-		ctl, exist := svr.ctlManager.GetByID(newMsg.RunID)
-		if !exist {
+		admitted, err := svr.ctlManager.admitVisitorByRunID(newMsg.RunID, admit)
+		if err != nil {
+			return err
+		}
+		if !admitted {
 			return fmt.Errorf("no client control found for run id [%s]", newMsg.RunID)
 		}
-		visitorUser = ctl.sessionCtx.LoginMsg.User
+		return nil
 	}
-	return svr.rc.VisitorManager.NewConn(newMsg.ProxyName, visitorConn, newMsg.Timestamp, newMsg.SignKey,
-		newMsg.UseEncryption, newMsg.UseCompression, visitorUser, wireProtocol)
+	return admit("")
 }

server/service_test.go

@@ -15,12 +15,27 @@
 package server
 
 import (
+	"context"
 	"errors"
 	"net"
+	"runtime"
+	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/require"
+
+	"github.com/fatedier/frp/pkg/auth"
+	v1 "github.com/fatedier/frp/pkg/config/v1"
+	"github.com/fatedier/frp/pkg/msg"
+	plugin "github.com/fatedier/frp/pkg/plugin/server"
+	"github.com/fatedier/frp/pkg/proto/wire"
+	"github.com/fatedier/frp/pkg/util/util"
+	"github.com/fatedier/frp/server/controller"
+	"github.com/fatedier/frp/server/proxy"
+	"github.com/fatedier/frp/server/registry"
+	"github.com/fatedier/frp/server/visitor"
 )
 
 func TestWriteWithDeadlineTimesOutAndClearsDeadline(t *testing.T) {
@@ -61,3 +76,504 @@ func TestWriteWithDeadlineTimesOutAndClearsDeadline(t *testing.T) {
 		t.Fatal("timed out waiting for write after deadline reset")
 	}
 }
+
+func TestServiceControlHandoffSkipsStalePendingGeneration(t *testing.T) {
+	svr := newControlTestService(t)
+	metrics := newCountingServerMetrics()
+	metrics.closeEnter = make(chan struct{})
+	metrics.closeResume = make(chan struct{})
+
+	ctlA, connA, err := registerLifecycleTestControl(svr)
+	require.NoError(t, err)
+	ctlA.serverMetrics = metrics
+	require.NoError(t, svr.completeControlLogin(ctlA, func() error { return nil }))
+	waitForSignal(t, connA.readStarted, "A reader to start")
+
+	require.NoError(t, ctlA.Close())
+	waitForSignal(t, metrics.closeEnter, "A finalization barrier")
+
+	type registerResult struct {
+		ctl  *Control
+		conn *deadlineReadConn
+		err  error
+	}
+	resultB := make(chan registerResult, 1)
+	go func() {
+		ctl, conn, registerErr := registerLifecycleTestControl(svr)
+		resultB <- registerResult{ctl: ctl, conn: conn, err: registerErr}
+	}()
+	ctlB := waitForDifferentCurrentControl(t, svr.ctlManager, "shared-run", ctlA)
+	ctlB.serverMetrics = metrics
+
+	resultC := make(chan registerResult, 1)
+	go func() {
+		ctl, conn, registerErr := registerLifecycleTestControl(svr)
+		resultC <- registerResult{ctl: ctl, conn: conn, err: registerErr}
+	}()
+	ctlC := waitForDifferentCurrentControl(t, svr.ctlManager, "shared-run", ctlB)
+	ctlC.serverMetrics = metrics
+	waitForControlDone(t, ctlB)
+
+	select {
+	case result := <-resultB:
+		t.Fatalf("B returned before A finalized: %v", result.err)
+	default:
+	}
+	select {
+	case result := <-resultC:
+		t.Fatalf("C returned before A finalized: %v", result.err)
+	default:
+	}
+
+	close(metrics.closeResume)
+	waitForControlDone(t, ctlA)
+
+	b := <-resultB
+	require.Same(t, ctlB, b.ctl)
+	require.ErrorIs(t, b.err, errControlReplaced)
+	require.False(t, svr.ctlManager.Remove(ctlB))
+	require.NoError(t, ctlB.Close())
+
+	c := <-resultC
+	require.NoError(t, c.err)
+	require.Same(t, ctlC, c.ctl)
+	_, ok := svr.ctlManager.GetByID("shared-run")
+	require.False(t, ok)
+	require.Same(t, ctlC, currentControlForTest(svr.ctlManager, "shared-run"))
+
+	info, ok := svr.clientRegistry.GetByKey("client")
+	require.True(t, ok)
+	require.True(t, info.Online)
+	require.Equal(t, uint64(ctlC.ID()), info.ControlID)
+
+	var staleWrites atomic.Int64
+	err = svr.completeControlLogin(ctlB, func() error {
+		staleWrites.Add(1)
+		return nil
+	})
+	require.ErrorIs(t, err, errControlReplaced)
+	require.Equal(t, int64(0), staleWrites.Load())
+
+	require.NoError(t, svr.completeControlLogin(ctlC, func() error { return nil }))
+	waitForSignal(t, c.conn.readStarted, "C reader to start")
+	current, ok := svr.ctlManager.GetByID("shared-run")
+	require.True(t, ok)
+	require.Same(t, ctlC, current)
+	require.Equal(t, int64(2), metrics.newClients())
+	require.Equal(t, int64(1), metrics.closedClients())
+
+	require.NoError(t, ctlC.Close())
+	waitForControlDone(t, ctlC)
+	require.Equal(t, int64(2), metrics.newClients())
+	require.Equal(t, int64(2), metrics.closedClients())
+	_, ok = svr.ctlManager.GetByID("shared-run")
+	require.False(t, ok)
+}
+
+func TestServiceLoginResponseSynchronizationIsScopedToRun(t *testing.T) {
+	svr := newControlTestService(t)
+	metrics := newCountingServerMetrics()
+	ctlA, connA, err := registerLifecycleTestControl(svr)
+	require.NoError(t, err)
+	ctlA.serverMetrics = metrics
+
+	writeEntered := make(chan struct{})
+	resumeWrite := make(chan struct{})
+	var resumeWriteOnce sync.Once
+	resume := func() {
+		resumeWriteOnce.Do(func() { close(resumeWrite) })
+	}
+	t.Cleanup(resume)
+	writeCount := atomic.Int64{}
+	loginDone := make(chan error, 1)
+	go func() {
+		loginDone <- svr.completeControlLogin(ctlA, func() error {
+			close(writeEntered)
+			<-resumeWrite
+			writeCount.Add(1)
+			return nil
+		})
+	}()
+	waitForSignal(t, writeEntered, "A LoginResp write")
+
+	runMu := currentRunGateForTest(svr.ctlManager, "shared-run")
+	require.NotNil(t, runMu)
+	if !svr.ctlManager.mu.TryLock() {
+		t.Fatal("ControlManager mutex was held while LoginResp write was in progress")
+	}
+	svr.ctlManager.mu.Unlock()
+
+	ctlB, connB := newLifecycleTestControl(t, "shared-run", "client", metrics)
+	gateAvailable := make(chan bool)
+	addDone := make(chan error, 1)
+	go func() {
+		if runMu.TryLock() {
+			runMu.Unlock()
+			gateAvailable <- true
+		} else {
+			gateAvailable <- false
+		}
+		addErr := svr.ctlManager.Add(ctlB)
+		addDone <- addErr
+	}()
+	available := waitForResult(t, gateAvailable, "same-run replacement gate probe")
+	require.False(t, available, "same-run gate was available to replacement during LoginResp write")
+	select {
+	case addErr := <-addDone:
+		t.Fatalf("same-run replacement completed during LoginResp write: %v", addErr)
+	case <-time.After(20 * time.Millisecond):
+	}
+	require.Same(t, ctlA, currentControlForTest(svr.ctlManager, "shared-run"))
+
+	otherMetrics := newCountingServerMetrics()
+	otherCtl, otherConn := newLifecycleTestControl(t, "other-run", "other-client", otherMetrics)
+	type unrelatedResult struct {
+		addErr      error
+		active      bool
+		activateErr error
+		loginErr    error
+		current     *Control
+		found       bool
+	}
+	unrelatedDone := make(chan unrelatedResult, 1)
+	go func() {
+		result := unrelatedResult{}
+		result.addErr = svr.ctlManager.Add(otherCtl)
+		if result.addErr == nil {
+			result.active, result.activateErr = svr.ctlManager.Activate(otherCtl)
+		}
+		if result.activateErr == nil && result.active {
+			result.loginErr = svr.completeControlLogin(otherCtl, func() error { return nil })
+		}
+		result.current, result.found = svr.ctlManager.GetByID("other-run")
+		unrelatedDone <- result
+	}()
+	result := waitForResult(t, unrelatedDone, "unrelated run lifecycle")
+	require.NoError(t, result.addErr)
+	require.NoError(t, result.activateErr)
+	require.True(t, result.active)
+	require.NoError(t, result.loginErr)
+	require.True(t, result.found)
+	require.Same(t, otherCtl, result.current)
+	waitForSignal(t, otherConn.readStarted, "unrelated control reader to start")
+	require.Equal(t, int64(1), otherMetrics.newClients())
+
+	resume()
+	require.NoError(t, waitForResult(t, loginDone, "LoginResp completion"))
+	require.NoError(t, waitForResult(t, addDone, "replacement"))
+	waitForControlDone(t, ctlA)
+	require.Same(t, ctlB, currentControlForTest(svr.ctlManager, "shared-run"))
+	require.Equal(t, int64(1), writeCount.Load())
+	require.Equal(t, int64(1), metrics.newClients())
+	require.Equal(t, int64(1), metrics.closedClients())
+	require.Equal(t, []string{"deadline", "close"}, connA.eventsSnapshot())
+
+	require.False(t, svr.ctlManager.Remove(ctlA))
+	require.NoError(t, ctlA.Close())
+	require.True(t, svr.ctlManager.Remove(ctlB))
+	require.NoError(t, ctlB.Close())
+	require.Equal(t, []string{"deadline", "close"}, connB.eventsSnapshot())
+
+	require.NoError(t, otherCtl.Close())
+	waitForControlDone(t, otherCtl)
+	require.Equal(t, int64(1), otherMetrics.newClients())
+	require.Equal(t, int64(1), otherMetrics.closedClients())
+}
+
+func TestServiceVisitorAdmissionSerializesReplacement(t *testing.T) {
+	svr := newControlTestService(t)
+	ctlA, controlConn, err := registerLifecycleTestControl(svr)
+	require.NoError(t, err)
+	ctlA.sessionCtx.LoginMsg.User = "old-user"
+	require.NoError(t, svr.completeControlLogin(ctlA, func() error { return nil }))
+	waitForSignal(t, controlConn.readStarted, "A reader to start")
+
+	admissionEntered := make(chan struct{})
+	resumeAdmission := make(chan struct{})
+	var resumeOnce sync.Once
+	resume := func() {
+		resumeOnce.Do(func() { close(resumeAdmission) })
+	}
+	t.Cleanup(resume)
+	type admissionResult struct {
+		admitted bool
+		user     string
+		err      error
+	}
+	admissionDone := make(chan admissionResult, 1)
+	go func() {
+		var admittedUser string
+		admitted, admitErr := svr.ctlManager.admitVisitorByRunID("shared-run", func(user string) error {
+			admittedUser = user
+			close(admissionEntered)
+			<-resumeAdmission
+			return nil
+		})
+		admissionDone <- admissionResult{admitted: admitted, user: admittedUser, err: admitErr}
+	}()
+	waitForSignal(t, admissionEntered, "visitor admission callback")
+	runMu := currentRunGateForTest(svr.ctlManager, "shared-run")
+	require.NotNil(t, runMu)
+
+	type registerResult struct {
+		ctl *Control
+		err error
+	}
+	gateAvailable := make(chan bool)
+	replacementDone := make(chan registerResult, 1)
+	go func() {
+		if runMu.TryLock() {
+			runMu.Unlock()
+			gateAvailable <- true
+		} else {
+			gateAvailable <- false
+		}
+		ctl, _, registerErr := registerLifecycleTestControl(svr)
+		replacementDone <- registerResult{ctl: ctl, err: registerErr}
+	}()
+	available := waitForResult(t, gateAvailable, "visitor replacement gate probe")
+	require.False(t, available, "same-run gate was available during visitor admission")
+	select {
+	case result := <-replacementDone:
+		t.Fatalf("replacement completed during visitor admission: %v", result.err)
+	case <-time.After(20 * time.Millisecond):
+	}
+	require.Same(t, ctlA, currentControlForTest(svr.ctlManager, "shared-run"))
+
+	resume()
+	admission := waitForResult(t, admissionDone, "visitor admission")
+	require.NoError(t, admission.err)
+	require.True(t, admission.admitted)
+	require.Equal(t, "old-user", admission.user)
+	replacement := waitForResult(t, replacementDone, "replacement")
+	require.NoError(t, replacement.err)
+	ctlB := replacement.ctl
+	require.Same(t, ctlB, currentControlForTest(svr.ctlManager, "shared-run"))
+	waitForControlDone(t, ctlA)
+	require.True(t, svr.ctlManager.Remove(ctlB))
+	require.NoError(t, ctlB.Close())
+}
+
+func TestServiceWorkConnRoutingRequiresCurrentRunningControl(t *testing.T) {
+	svr := newControlTestService(t)
+	ctl, controlConn, err := registerLifecycleTestControl(svr)
+	require.NoError(t, err)
+
+	pendingConn := newCountingCloseConn()
+	pendingMsgConn := msg.NewConn(pendingConn, msg.NewV1ReadWriter(pendingConn))
+	err = registerWorkConnAsCaller(svr, pendingMsgConn, &msg.NewWorkConn{RunID: "shared-run"})
+	require.Error(t, err)
+	require.Equal(t, int64(1), pendingConn.closeCount.Load())
+	require.Len(t, ctl.workConnCh, 0)
+
+	require.NoError(t, svr.completeControlLogin(ctl, func() error { return nil }))
+	waitForSignal(t, controlConn.readStarted, "control reader to start")
+	current, ok := svr.ctlManager.GetByID("shared-run")
+	require.True(t, ok)
+	require.Same(t, ctl, current)
+	require.Len(t, ctl.workConnCh, 0)
+
+	runningConn := newCountingCloseConn()
+	runningMsgConn := msg.NewConn(runningConn, msg.NewV1ReadWriter(runningConn))
+	require.NoError(t, svr.RegisterWorkConn(runningMsgConn, &msg.NewWorkConn{RunID: "shared-run"}))
+	require.Len(t, ctl.workConnCh, 1)
+
+	require.NoError(t, ctl.Close())
+	waitForControlDone(t, ctl)
+	require.Equal(t, int64(1), runningConn.closeCount.Load())
+}
+
+func TestServiceWorkConnRoutingRejectsLostGeneration(t *testing.T) {
+	for _, action := range []string{"replace", "close"} {
+		t.Run(action, func(t *testing.T) {
+			svr := newControlTestService(t)
+			ctl, controlConn, err := registerLifecycleTestControl(svr)
+			require.NoError(t, err)
+			require.NoError(t, svr.completeControlLogin(ctl, func() error { return nil }))
+			waitForSignal(t, controlConn.readStarted, "control reader to start")
+
+			barrier := newWorkConnBarrierPlugin()
+			svr.pluginManager.Register(barrier)
+			workConn := newCountingCloseConn()
+			workMsgConn := msg.NewConn(workConn, msg.NewV1ReadWriter(workConn))
+			routeDone := make(chan error, 1)
+			go func() {
+				routeDone <- registerWorkConnAsCaller(svr, workMsgConn, &msg.NewWorkConn{RunID: "shared-run"})
+			}()
+			waitForSignal(t, barrier.entered, "work connection plugin barrier")
+
+			var replacement *Control
+			switch action {
+			case "replace":
+				replacement, _, err = registerLifecycleTestControl(svr)
+				require.NoError(t, err)
+			case "close":
+				require.NoError(t, ctl.Close())
+				waitForControlDone(t, ctl)
+			}
+
+			close(barrier.resume)
+			require.Error(t, waitForResult(t, routeDone, "work connection route to finish"))
+			require.Equal(t, int64(1), workConn.closeCount.Load())
+			require.Len(t, ctl.workConnCh, 0)
+
+			if replacement != nil {
+				require.Len(t, replacement.workConnCh, 0)
+				require.True(t, svr.ctlManager.Remove(replacement))
+				require.NoError(t, replacement.Close())
+				waitForControlDone(t, replacement)
+			}
+		})
+	}
+}
+
+func TestServiceVisitorRoutingExcludesPendingUser(t *testing.T) {
+	svr := newControlTestService(t)
+	listener, err := svr.rc.VisitorManager.Listen("visitor", "secret", []string{"pending-user"})
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = listener.Close() })
+
+	controlConn := newDeadlineReadConn()
+	controlMsgConn := msg.NewConn(controlConn, msg.NewV1ReadWriter(controlConn))
+	ctl, err := svr.RegisterControl(controlMsgConn, &msg.Login{
+		RunID:    "visitor-run",
+		User:     "pending-user",
+		ClientID: "visitor-client",
+		ClientSpec: msg.ClientSpec{
+			AlwaysAuthPass: true,
+		},
+	}, true, wire.ProtocolV1)
+	require.NoError(t, err)
+
+	timestamp := time.Now().Unix()
+	visitorMsg := &msg.NewVisitorConn{
+		RunID:     "visitor-run",
+		ProxyName: "visitor",
+		Timestamp: timestamp,
+		SignKey:   util.GetAuthKey("secret", timestamp),
+	}
+	pendingConn := newCountingCloseConn()
+	err = svr.RegisterVisitorConn(pendingConn, visitorMsg, wire.ProtocolV1)
+	require.ErrorContains(t, err, "no client control found")
+	require.NoError(t, pendingConn.Close())
+	require.Equal(t, int64(1), pendingConn.closeCount.Load())
+
+	require.NoError(t, svr.completeControlLogin(ctl, func() error { return nil }))
+	waitForSignal(t, controlConn.readStarted, "control reader to start")
+	runningConn := newCountingCloseConn()
+	require.NoError(t, svr.RegisterVisitorConn(runningConn, visitorMsg, wire.ProtocolV1))
+	accepted, err := listener.Accept()
+	require.NoError(t, err)
+	require.NoError(t, accepted.Close())
+	require.Equal(t, int64(1), runningConn.closeCount.Load())
+
+	require.NoError(t, ctl.Close())
+	waitForControlDone(t, ctl)
+}
+
+func newControlTestService(t *testing.T) *Service {
+	t.Helper()
+	cfg := &v1.ServerConfig{}
+	cfg.Auth.Method = v1.AuthMethodToken
+	authRuntime, err := auth.BuildServerAuth(&cfg.Auth)
+	require.NoError(t, err)
+	clientRegistry := registry.NewClientRegistry()
+	return &Service{
+		ctlManager:     NewControlManager(clientRegistry),
+		clientRegistry: clientRegistry,
+		pxyManager:     proxy.NewManager(),
+		pluginManager:  plugin.NewManager(),
+		rc: &controller.ResourceController{
+			VisitorManager: visitor.NewManager(),
+		},
+		auth: authRuntime,
+		cfg:  cfg,
+	}
+}
+
+func registerLifecycleTestControl(svr *Service) (*Control, *deadlineReadConn, error) {
+	conn := newDeadlineReadConn()
+	msgConn := msg.NewConn(conn, msg.NewReadWriter(conn, wire.ProtocolV1))
+	ctl, err := svr.RegisterControl(msgConn, &msg.Login{
+		RunID:    "shared-run",
+		ClientID: "client",
+		ClientSpec: msg.ClientSpec{
+			AlwaysAuthPass: true,
+		},
+	}, true, wire.ProtocolV1)
+	return ctl, conn, err
+}
+
+func waitForDifferentCurrentControl(t *testing.T, manager *ControlManager, runID string, old *Control) *Control {
+	t.Helper()
+	deadline := time.Now().Add(3 * time.Second)
+	for time.Now().Before(deadline) {
+		if ctl := currentControlForTest(manager, runID); ctl != nil && ctl != old {
+			return ctl
+		}
+		runtime.Gosched()
+	}
+	t.Fatalf("timed out waiting for a new current control after ID %d", old.ID())
+	return nil
+}
+
+func registerWorkConnAsCaller(svr *Service, workConn *msg.Conn, newMsg *msg.NewWorkConn) error {
+	err := svr.RegisterWorkConn(workConn, newMsg)
+	if err != nil {
+		_ = workConn.Close()
+	}
+	return err
+}
+
+func waitForResult[T any](t *testing.T, ch <-chan T, description string) T {
+	t.Helper()
+	select {
+	case result := <-ch:
+		return result
+	case <-time.After(3 * time.Second):
+		t.Fatalf("timed out waiting for %s", description)
+		var zero T
+		return zero
+	}
+}
+
+type workConnBarrierPlugin struct {
+	entered chan struct{}
+	resume  chan struct{}
+}
+
+func newWorkConnBarrierPlugin() *workConnBarrierPlugin {
+	return &workConnBarrierPlugin{
+		entered: make(chan struct{}),
+		resume:  make(chan struct{}),
+	}
+}
+
+func (*workConnBarrierPlugin) Name() string { return "work-conn-barrier" }
+
+func (*workConnBarrierPlugin) IsSupport(op string) bool { return op == plugin.OpNewWorkConn }
+
+func (p *workConnBarrierPlugin) Handle(
+	context.Context,
+	string,
+	any,
+) (*plugin.Response, any, error) {
+	close(p.entered)
+	<-p.resume
+	return &plugin.Response{Unchange: true}, nil, nil
+}
+
+type countingCloseConn struct {
+	closeCount atomic.Int64
+}
+
+func newCountingCloseConn() *countingCloseConn { return &countingCloseConn{} }
+
+func (*countingCloseConn) Read([]byte) (int, error)         { return 0, net.ErrClosed }
+func (*countingCloseConn) Write(p []byte) (int, error)      { return len(p), nil }
+func (c *countingCloseConn) Close() error                   { c.closeCount.Add(1); return nil }
+func (*countingCloseConn) LocalAddr() net.Addr              { return lifecycleTestAddr("local") }
+func (*countingCloseConn) RemoteAddr() net.Addr             { return lifecycleTestAddr("remote") }
+func (*countingCloseConn) SetDeadline(time.Time) error      { return nil }
+func (*countingCloseConn) SetReadDeadline(time.Time) error  { return nil }
+func (*countingCloseConn) SetWriteDeadline(time.Time) error { return nil }

test/e2e/framework/process.go

@@ -94,11 +94,9 @@ func (f *Framework) RunProcessesWithBinaries(
 }
 
 func (f *Framework) RunFrps(args ...string) (*process.Process, string, error) {
-	p := process.NewWithEnvs(TestContext.FRPServerPath, args, f.osEnvs)
-	f.serverProcesses = append(f.serverProcesses, p)
-	err := p.Start()
+	p, output, err := f.StartFrps(args...)
 	if err != nil {
-		return p, p.Output(), err
+		return p, output, err
 	}
 	select {
 	case <-p.Done():
@@ -107,17 +105,39 @@ func (f *Framework) RunFrps(args ...string) (*process.Process, string, error) {
 	return p, p.Output(), nil
 }
 
+// StartFrps starts frps without an implicit sleep so tests can wait on an
+// explicit readiness event.
+func (f *Framework) StartFrps(args ...string) (*process.Process, string, error) {
+	p := process.NewWithEnvs(TestContext.FRPServerPath, args, f.osEnvs)
+	f.serverProcesses = append(f.serverProcesses, p)
+	err := p.Start()
+	if err != nil {
+		return p, p.Output(), err
+	}
+	return p, p.Output(), nil
+}
+
 func (f *Framework) RunFrpc(args ...string) (*process.Process, string, error) {
+	p, output, err := f.StartFrpc(args...)
+	if err != nil {
+		return p, output, err
+	}
+	select {
+	case <-p.Done():
+	case <-time.After(1500 * time.Millisecond):
+	}
+	return p, p.Output(), nil
+}
+
+// StartFrpc starts frpc without an implicit sleep so tests can wait on an
+// explicit login or proxy-readiness event.
+func (f *Framework) StartFrpc(args ...string) (*process.Process, string, error) {
 	p := process.NewWithEnvs(TestContext.FRPClientPath, args, f.osEnvs)
 	f.clientProcesses = append(f.clientProcesses, p)
 	err := p.Start()
 	if err != nil {
 		return p, p.Output(), err
 	}
-	select {
-	case <-p.Done():
-	case <-time.After(1500 * time.Millisecond):
-	}
 	return p, p.Output(), nil
 }
 

test/e2e/pkg/relay/halfopen.go

@@ -0,0 +1,208 @@
+// Copyright 2026 The frp Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package relay
+
+import (
+	"fmt"
+	"io"
+	"net"
+	"strconv"
+	"sync"
+	"time"
+)
+
+// HalfOpen forwards TCP connections until Blackhole is called. A blackholed
+// pair stops forwarding but deliberately retains the upstream socket so the
+// peer sees a real half-open connection until the relay is closed.
+type HalfOpen struct {
+	bindAddr     string
+	bindPort     int
+	upstreamAddr string
+
+	listener net.Listener
+	done     chan struct{}
+	accepted chan struct{}
+
+	mu    sync.Mutex
+	pairs []*connectionPair
+
+	wg        sync.WaitGroup
+	closeOnce sync.Once
+}
+
+type connectionPair struct {
+	downstream net.Conn
+	upstream   net.Conn
+
+	mu         sync.Mutex
+	blackholed bool
+}
+
+func New(upstreamAddr string) *HalfOpen {
+	return &HalfOpen{
+		bindAddr:     "127.0.0.1",
+		upstreamAddr: upstreamAddr,
+		done:         make(chan struct{}),
+		accepted:     make(chan struct{}, 1),
+	}
+}
+
+func (r *HalfOpen) Run() error {
+	listener, err := net.Listen("tcp", net.JoinHostPort(r.bindAddr, strconv.Itoa(r.bindPort)))
+	if err != nil {
+		return err
+	}
+	r.listener = listener
+	r.bindPort = listener.Addr().(*net.TCPAddr).Port
+
+	r.wg.Add(1)
+	go r.acceptLoop()
+	return nil
+}
+
+func (r *HalfOpen) acceptLoop() {
+	defer r.wg.Done()
+	for {
+		downstream, err := r.listener.Accept()
+		if err != nil {
+			return
+		}
+		upstream, err := net.DialTimeout("tcp", r.upstreamAddr, 3*time.Second)
+		if err != nil {
+			_ = downstream.Close()
+			continue
+		}
+
+		pair := &connectionPair{downstream: downstream, upstream: upstream}
+		r.mu.Lock()
+		r.pairs = append(r.pairs, pair)
+		r.mu.Unlock()
+		select {
+		case r.accepted <- struct{}{}:
+		default:
+		}
+
+		r.wg.Add(1)
+		go r.servePair(pair)
+	}
+}
+
+func (r *HalfOpen) servePair(pair *connectionPair) {
+	defer r.wg.Done()
+	copyDone := make(chan struct{}, 2)
+	go func() {
+		_, _ = io.Copy(pair.upstream, pair.downstream)
+		copyDone <- struct{}{}
+	}()
+	go func() {
+		_, _ = io.Copy(pair.downstream, pair.upstream)
+		copyDone <- struct{}{}
+	}()
+
+	completed := 0
+	select {
+	case <-copyDone:
+		completed = 1
+		if !pair.isBlackholed() {
+			_ = pair.downstream.Close()
+			_ = pair.upstream.Close()
+		}
+	case <-r.done:
+		_ = pair.downstream.Close()
+		_ = pair.upstream.Close()
+	}
+	for completed < 2 {
+		<-copyDone
+		completed++
+	}
+
+	if pair.isBlackholed() {
+		<-r.done
+		_ = pair.downstream.Close()
+		_ = pair.upstream.Close()
+	}
+}
+
+func (r *HalfOpen) WaitForConnections(count int, timeout time.Duration) error {
+	timer := time.NewTimer(timeout)
+	defer timer.Stop()
+	for {
+		r.mu.Lock()
+		accepted := len(r.pairs)
+		r.mu.Unlock()
+		if accepted >= count {
+			return nil
+		}
+		select {
+		case <-r.accepted:
+		case <-r.done:
+			return fmt.Errorf("relay closed after accepting %d of %d connections", accepted, count)
+		case <-timer.C:
+			return fmt.Errorf("timed out after accepting %d of %d connections", accepted, count)
+		}
+	}
+}
+
+// Blackhole uses a one-based connection index in accept order.
+func (r *HalfOpen) Blackhole(index int) error {
+	r.mu.Lock()
+	if index <= 0 || index > len(r.pairs) {
+		accepted := len(r.pairs)
+		r.mu.Unlock()
+		return fmt.Errorf("connection %d is unavailable; accepted %d", index, accepted)
+	}
+	pair := r.pairs[index-1]
+	r.mu.Unlock()
+
+	pair.mu.Lock()
+	if pair.blackholed {
+		pair.mu.Unlock()
+		return nil
+	}
+	pair.blackholed = true
+	pair.mu.Unlock()
+
+	now := time.Now()
+	_ = pair.downstream.SetDeadline(now)
+	_ = pair.upstream.SetDeadline(now)
+	return nil
+}
+
+func (r *HalfOpen) Close() error {
+	r.closeOnce.Do(func() {
+		close(r.done)
+		if r.listener != nil {
+			_ = r.listener.Close()
+		}
+		r.mu.Lock()
+		pairs := append([]*connectionPair(nil), r.pairs...)
+		r.mu.Unlock()
+		for _, pair := range pairs {
+			_ = pair.downstream.Close()
+			_ = pair.upstream.Close()
+		}
+		r.wg.Wait()
+	})
+	return nil
+}
+
+func (r *HalfOpen) BindAddr() string { return r.bindAddr }
+func (r *HalfOpen) BindPort() int    { return r.bindPort }
+
+func (p *connectionPair) isBlackholed() bool {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	return p.blackholed
+}

test/e2e/v1/features/control_replacement.go

@@ -0,0 +1,300 @@
+// Copyright 2026 The frp Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package features
+
+import (
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/onsi/ginkgo/v2"
+
+	"github.com/fatedier/frp/test/e2e/framework"
+	"github.com/fatedier/frp/test/e2e/framework/consts"
+	"github.com/fatedier/frp/test/e2e/pkg/relay"
+	"github.com/fatedier/frp/test/e2e/pkg/request"
+)
+
+var _ = ginkgo.Describe("[Feature: ControlReplacement]", func() {
+	f := framework.NewDefaultFramework()
+
+	for _, wireProtocol := range []string{"v1", "v2"} {
+		for _, tcpMux := range []bool{true, false} {
+			ginkgo.It(fmt.Sprintf("recovers a %s control through a half-open relay with tcpMux=%t", wireProtocol, tcpMux), func() {
+				runHalfOpenControlReplacement(f, wireProtocol, tcpMux)
+			})
+		}
+	}
+})
+
+func runHalfOpenControlReplacement(f *framework.Framework, wireProtocol string, tcpMux bool) {
+	serverPort := f.AllocPort()
+	dashboardPort := f.AllocPort()
+	remotePort := f.AllocPort()
+	heartbeatTimeout := int64(-1)
+	if !tcpMux {
+		heartbeatTimeout = 3
+	}
+
+	serverConfig := fmt.Sprintf(`
+bindAddr = "127.0.0.1"
+bindPort = %d
+log.level = "trace"
+transport.tcpMux = %t
+transport.tcpMuxKeepaliveInterval = 30
+transport.heartbeatTimeout = %d
+webServer.addr = "127.0.0.1"
+webServer.port = %d
+webServer.pprofEnable = true
+enablePrometheus = true
+`, serverPort, tcpMux, heartbeatTimeout, dashboardPort)
+	serverConfigPath := f.WriteTempFile("issue-5391-frps.toml", serverConfig)
+	serverProcess, _, err := f.StartFrps("-c", serverConfigPath)
+	framework.ExpectNoError(err)
+	framework.ExpectNoError(framework.WaitForTCPReady(fmt.Sprintf("127.0.0.1:%d", serverPort), 5*time.Second))
+
+	halfOpenRelay := relay.New(fmt.Sprintf("127.0.0.1:%d", serverPort))
+	f.RunServer("", halfOpenRelay)
+
+	heartbeatInterval := int64(-1)
+	clientHeartbeatTimeout := int64(-1)
+	if !tcpMux {
+		heartbeatInterval = 1
+		clientHeartbeatTimeout = 3
+	}
+	clientConfig := fmt.Sprintf(`
+serverAddr = "127.0.0.1"
+serverPort = %d
+clientID = "issue-5391"
+loginFailExit = false
+log.level = "trace"
+transport.wireProtocol = %q
+transport.tcpMux = %t
+transport.tcpMuxKeepaliveInterval = 1
+transport.heartbeatInterval = %d
+transport.heartbeatTimeout = %d
+transport.tls.enable = false
+
+[[proxies]]
+name = "issue-5391-tcp"
+type = "tcp"
+localPort = %d
+remotePort = %d
+`, halfOpenRelay.BindPort(), wireProtocol, tcpMux, heartbeatInterval, clientHeartbeatTimeout,
+		f.PortByName(framework.TCPEchoServerPort), remotePort)
+	clientConfigPath := f.WriteTempFile("issue-5391-frpc.toml", clientConfig)
+	clientProcess, _, err := f.StartFrpc("-c", clientConfigPath)
+	framework.ExpectNoError(err)
+	framework.ExpectNoError(halfOpenRelay.WaitForConnections(1, 5*time.Second))
+	framework.ExpectNoError(clientProcess.WaitForOutput("login to server success", 1, 10*time.Second))
+	framework.ExpectNoError(clientProcess.WaitForOutput("[issue-5391-tcp] start proxy success", 1, 10*time.Second))
+	framework.ExpectNoError(waitForReplacementState(dashboardPort, remotePort, 10*time.Second))
+
+	replacementCount := 1
+	if tcpMux {
+		replacementCount = 3
+	}
+	for i := 0; i < replacementCount; i++ {
+		connectionIndex := 1
+		if tcpMux {
+			connectionIndex = i + 1
+		}
+		framework.ExpectNoError(halfOpenRelay.Blackhole(connectionIndex))
+		if tcpMux {
+			framework.ExpectNoError(halfOpenRelay.WaitForConnections(connectionIndex+1, 15*time.Second))
+		}
+		framework.ExpectNoError(clientProcess.WaitForOutput("login to server success", i+2, 15*time.Second))
+		framework.ExpectNoError(clientProcess.WaitForOutput("[issue-5391-tcp] start proxy success", i+2, 15*time.Second))
+		framework.ExpectNoError(waitForReplacementState(dashboardPort, remotePort, 10*time.Second))
+	}
+
+	_ = clientProcess.Stop()
+	select {
+	case <-clientProcess.Done():
+	case <-time.After(5 * time.Second):
+		framework.Failf("frpc did not exit")
+	}
+	framework.ExpectNoError(waitForReplacementShutdown(dashboardPort, 10*time.Second))
+	framework.ExpectNoError(halfOpenRelay.Close())
+	framework.ExpectNoError(waitForNoHandoffWaiters(dashboardPort, 5*time.Second))
+
+	_ = serverProcess.Stop()
+	select {
+	case <-serverProcess.Done():
+	case <-time.After(5 * time.Second):
+		framework.Failf("frps did not exit")
+	}
+}
+
+func waitForReplacementState(dashboardPort, remotePort int, timeout time.Duration) error {
+	return waitForLifecycleCondition(timeout, func() error {
+		metricsBody, err := getLifecycleEndpoint(dashboardPort, "/metrics")
+		if err != nil {
+			return err
+		}
+		if err := expectMetricValue(metricsBody, "frp_server_client_counts", "", 1); err != nil {
+			return err
+		}
+		if err := expectMetricValue(metricsBody, "frp_server_proxy_counts", `type="tcp"`, 1); err != nil {
+			return err
+		}
+		clients, err := getOnlineLifecycleClients(dashboardPort)
+		if err != nil {
+			return err
+		}
+		if len(clients) != 1 || clients[0].ClientID != "issue-5391" {
+			return fmt.Errorf("expected one online client, got %+v", clients)
+		}
+		profile, err := getLifecycleEndpoint(dashboardPort, "/debug/pprof/goroutine?debug=2")
+		if err != nil {
+			return err
+		}
+		if err := expectNoHandoffWaiter(profile, "after replacement"); err != nil {
+			return err
+		}
+		resp, err := request.New().
+			TCP().
+			Port(remotePort).
+			Timeout(time.Second).
+			Body([]byte(consts.TestString)).
+			Do()
+		if err != nil {
+			return err
+		}
+		if string(resp.Content) != consts.TestString {
+			return fmt.Errorf("unexpected proxy response %q", resp.Content)
+		}
+		return nil
+	})
+}
+
+func waitForReplacementShutdown(dashboardPort int, timeout time.Duration) error {
+	return waitForLifecycleCondition(timeout, func() error {
+		metricsBody, err := getLifecycleEndpoint(dashboardPort, "/metrics")
+		if err != nil {
+			return err
+		}
+		if err := expectMetricValue(metricsBody, "frp_server_client_counts", "", 0); err != nil {
+			return err
+		}
+		if err := expectMetricValue(metricsBody, "frp_server_proxy_counts", `type="tcp"`, 0); err != nil {
+			return err
+		}
+		clients, err := getOnlineLifecycleClients(dashboardPort)
+		if err != nil {
+			return err
+		}
+		if len(clients) != 0 {
+			return fmt.Errorf("expected no online clients, got %+v", clients)
+		}
+		return nil
+	})
+}
+
+func waitForNoHandoffWaiters(dashboardPort int, timeout time.Duration) error {
+	return waitForLifecycleCondition(timeout, func() error {
+		profile, err := getLifecycleEndpoint(dashboardPort, "/debug/pprof/goroutine?debug=2")
+		if err != nil {
+			return err
+		}
+		return expectNoHandoffWaiter(profile, "after relay shutdown")
+	})
+}
+
+type lifecycleClient struct {
+	ClientID string `json:"clientID"`
+}
+
+func expectNoHandoffWaiter(profile, phase string) error {
+	if strings.Contains(profile, "(*Control).WaitForHandoff") {
+		return fmt.Errorf("control handoff waiter remained %s", phase)
+	}
+	return nil
+}
+
+func getOnlineLifecycleClients(dashboardPort int) ([]lifecycleClient, error) {
+	body, err := getLifecycleEndpoint(dashboardPort, "/api/clients?status=online")
+	if err != nil {
+		return nil, err
+	}
+	var clients []lifecycleClient
+	if err := json.Unmarshal([]byte(body), &clients); err != nil {
+		return nil, err
+	}
+	return clients, nil
+}
+
+func getLifecycleEndpoint(port int, path string) (string, error) {
+	client := &http.Client{Timeout: time.Second}
+	resp, err := client.Get(fmt.Sprintf("http://127.0.0.1:%d%s", port, path))
+	if err != nil {
+		return "", err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		return "", fmt.Errorf("GET %s returned %s", path, resp.Status)
+	}
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return "", err
+	}
+	return string(body), nil
+}
+
+func expectMetricValue(body, name, labels string, want float64) error {
+	prefix := name
+	if labels != "" {
+		prefix += "{" + labels + "}"
+	}
+	for line := range strings.SplitSeq(body, "\n") {
+		fields := strings.Fields(line)
+		if len(fields) != 2 || fields[0] != prefix {
+			continue
+		}
+		got, err := strconv.ParseFloat(fields[1], 64)
+		if err != nil {
+			return err
+		}
+		if got != want {
+			return fmt.Errorf("metric %s = %v, want %v", prefix, got, want)
+		}
+		return nil
+	}
+	return fmt.Errorf("metric %s not found", prefix)
+}
+
+func waitForLifecycleCondition(timeout time.Duration, condition func() error) error {
+	timer := time.NewTimer(timeout)
+	defer timer.Stop()
+	ticker := time.NewTicker(25 * time.Millisecond)
+	defer ticker.Stop()
+	var lastErr error
+	for {
+		err := condition()
+		if err == nil {
+			return nil
+		}
+		lastErr = err
+		select {
+		case <-ticker.C:
+		case <-timer.C:
+			return fmt.Errorf("condition was not met: %w", lastErr)
+		}
+	}
+}

web/frps/src/api/proxy.ts

@@ -1,10 +1,14 @@
 import { buildQueryString, http } from './http'
+import { formatUnixSeconds } from '../utils/format'
 import type { V2Page } from './http'
 import type {
   GetProxyResponse,
   ProxyListV2Params,
   ProxyStatsInfo,
   ProxyV2Info,
+  ProxyV2Spec,
+  ProxyV2SpecBlocks,
+  ProxyV2Type,
   TrafficResponse,
 } from '../types/proxy'
 
@@ -38,19 +42,53 @@ export const getProxiesV2 = async (params: ProxyListV2Params = {}) => {
   }
 }
 
-export const toLegacyProxyStats = (proxy: ProxyV2Info): ProxyStatsInfo => ({
-  name: proxy.name,
-  type: proxy.type,
-  conf: proxy.spec,
-  user: proxy.user,
-  clientID: proxy.clientID,
-  todayTrafficIn: proxy.status.todayTrafficIn,
-  todayTrafficOut: proxy.status.todayTrafficOut,
-  curConns: proxy.status.curConns,
-  lastStartTime: proxy.status.lastStartTime,
-  lastCloseTime: proxy.status.lastCloseTime,
-  status: proxy.status.state || proxy.status.phase || '',
-})
+const getActiveProxySpec = (
+  spec: ProxyV2Spec,
+): ProxyV2SpecBlocks[ProxyV2Type] => {
+  switch (spec.type) {
+    case 'tcp':
+      return spec.tcp
+    case 'udp':
+      return spec.udp
+    case 'http':
+      return spec.http
+    case 'https':
+      return spec.https
+    case 'tcpmux':
+      return spec.tcpmux
+    case 'stcp':
+      return spec.stcp
+    case 'sudp':
+      return spec.sudp
+    case 'xtcp':
+      return spec.xtcp
+    default:
+      return assertNever(spec)
+  }
+}
+
+const assertNever = (value: never): never => {
+  throw new Error(`Unsupported proxy spec: ${JSON.stringify(value)}`)
+}
+
+export const toLegacyProxyStats = (proxy: ProxyV2Info): ProxyStatsInfo => {
+  const type = proxy.spec.type
+  const activeSpec = getActiveProxySpec(proxy.spec)
+
+  return {
+    name: proxy.name,
+    type,
+    conf: proxy.status.phase === 'offline' ? null : activeSpec,
+    user: proxy.user,
+    clientID: proxy.clientID,
+    todayTrafficIn: proxy.status.todayTrafficIn,
+    todayTrafficOut: proxy.status.todayTrafficOut,
+    curConns: proxy.status.curConns,
+    lastStartTime: formatUnixSeconds(proxy.status.lastStartAt),
+    lastCloseTime: formatUnixSeconds(proxy.status.lastCloseAt),
+    status: proxy.status.phase,
+  }
+}
 
 export const getProxy = (type: string, name: string) => {
   return http.get<ProxyStatsInfo>(`../api/proxy/${type}/${name}`)
@@ -68,7 +106,9 @@ export const getProxyByName = (name: string) => {
 }
 
 export const getProxyTraffic = (name: string) => {
-  return http.get<TrafficResponse>(`../api/traffic/${name}`)
+  return http.getV2<TrafficResponse>(
+    `../api/v2/proxies/${encodeURIComponent(name)}/traffic`,
+  )
 }
 
 export const clearOfflineProxies = () => {

web/frps/src/components/Traffic.vue

@@ -54,6 +54,7 @@ import { ref, onMounted } from 'vue'
 import { ElMessage } from 'element-plus'
 import { formatFileSize } from '../utils/format'
 import { getProxyTraffic } from '../api/proxy'
+import type { TrafficResponse } from '../types/proxy'
 
 const props = defineProps<{
   proxyName: string
@@ -71,41 +72,24 @@ const chartData = ref<
 >([])
 const maxVal = ref(0)
 
-const processData = (trafficIn: number[], trafficOut: number[]) => {
-  // Ensure we have arrays and reverse them (server returns newest first)
-  const inArr = [...(trafficIn || [])].reverse()
-  const outArr = [...(trafficOut || [])].reverse()
+const formatDateLabel = (date: string) => {
+  const parts = date.split('-')
+  if (parts.length !== 3) return date
+  return `${Number(parts[1])}-${Number(parts[2])}`
+}
 
-  // Pad with zeros if less than 7 days
-  while (inArr.length < 7) inArr.unshift(0)
-  while (outArr.length < 7) outArr.unshift(0)
-
-  // Slice to last 7 entries just in case
-  const finalIn = inArr.slice(-7)
-  const finalOut = outArr.slice(-7)
-
-  // Calculate dates (last 7 days ending today)
-  const dates: string[] = []
-  const d = new Date()
-  d.setDate(d.getDate() - 6)
-
-  for (let i = 0; i < 7; i++) {
-    dates.push(`${d.getMonth() + 1}-${d.getDate()}`)
-    d.setDate(d.getDate() + 1)
-  }
-
-  // Find max value for scaling
-  const maxIn = Math.max(...finalIn)
-  const maxOut = Math.max(...finalOut)
+const processData = (history: TrafficResponse['history'] = []) => {
+  const points = history || []
+  const maxIn = Math.max(0, ...points.map((item) => item.trafficIn))
+  const maxOut = Math.max(0, ...points.map((item) => item.trafficOut))
   maxVal.value = Math.max(maxIn, maxOut, 100) // Minimum scale 100 bytes
 
-  // Build chart data
-  chartData.value = dates.map((date, i) => ({
-    date,
-    in: finalIn[i],
-    out: finalOut[i],
-    inPercent: (finalIn[i] / maxVal.value) * 100,
-    outPercent: (finalOut[i] / maxVal.value) * 100,
+  chartData.value = points.map((item) => ({
+    date: formatDateLabel(item.date),
+    in: item.trafficIn,
+    out: item.trafficOut,
+    inPercent: (item.trafficIn / maxVal.value) * 100,
+    outPercent: (item.trafficOut / maxVal.value) * 100,
   }))
 }
 
@@ -113,7 +97,7 @@ const fetchData = () => {
   loading.value = true
   getProxyTraffic(props.proxyName)
     .then((json) => {
-      processData(json.trafficIn, json.trafficOut)
+      processData(json.history)
     })
     .catch((err) => {
       ElMessage({

web/frps/src/types/client.ts

@@ -7,7 +7,6 @@ export interface ClientInfoData {
   wireProtocol?: string
   hostname: string
   clientIP?: string
-  metas?: Record<string, string>
   firstConnectedAt: number
   lastConnectedAt: number
   disconnectedAt?: number

web/frps/src/types/proxy.ts

@@ -28,25 +28,102 @@ export interface ProxyListV2Params {
 
 export interface ProxyV2Info {
   name: string
-  type: string
   user: string
   clientID: string
-  spec: any
+  spec: ProxyV2Spec
   status: ProxyV2Status
 }
 
+export interface ProxyV2BaseSpec {
+  annotations?: Record<string, string>
+  metadatas?: Record<string, string>
+  transport?: {
+    useEncryption: boolean
+    useCompression: boolean
+    bandwidthLimit: string
+    bandwidthLimitMode: string
+  }
+  loadBalancer?: {
+    group: string
+  }
+}
+
+export interface ProxyV2TCPBlock extends ProxyV2BaseSpec {
+  remotePort?: number
+}
+
+export interface ProxyV2UDPBlock extends ProxyV2BaseSpec {
+  remotePort?: number
+}
+
+export interface ProxyV2HTTPBlock extends ProxyV2BaseSpec {
+  customDomains?: string[]
+  subdomain?: string
+  locations?: string[]
+  hostHeaderRewrite?: string
+}
+
+export interface ProxyV2HTTPSBlock extends ProxyV2BaseSpec {
+  customDomains?: string[]
+  subdomain?: string
+}
+
+export interface ProxyV2TCPMuxBlock extends ProxyV2BaseSpec {
+  customDomains?: string[]
+  subdomain?: string
+  multiplexer?: string
+  routeByHTTPUser?: string
+}
+
+export type ProxyV2STCPBlock = ProxyV2BaseSpec
+
+export type ProxyV2SUDPBlock = ProxyV2BaseSpec
+
+export type ProxyV2XTCPBlock = ProxyV2BaseSpec
+
+export interface ProxyV2SpecBlocks {
+  tcp: ProxyV2TCPBlock
+  udp: ProxyV2UDPBlock
+  http: ProxyV2HTTPBlock
+  https: ProxyV2HTTPSBlock
+  tcpmux: ProxyV2TCPMuxBlock
+  stcp: ProxyV2STCPBlock
+  sudp: ProxyV2SUDPBlock
+  xtcp: ProxyV2XTCPBlock
+}
+
+export type ProxyV2Type = keyof ProxyV2SpecBlocks
+
+type ProxyV2SpecFor<T extends ProxyV2Type> = {
+  type: T
+} & {
+  [K in T]: ProxyV2SpecBlocks[K]
+} & {
+  [K in Exclude<ProxyV2Type, T>]?: never
+}
+
+export type ProxyV2Spec = {
+  [T in ProxyV2Type]: ProxyV2SpecFor<T>
+}[ProxyV2Type]
+
 export interface ProxyV2Status {
-  state?: string
-  phase?: string
+  phase: 'online' | 'offline'
   todayTrafficIn: number
   todayTrafficOut: number
   curConns: number
-  lastStartTime: string
-  lastCloseTime: string
+  lastStartAt?: number
+  lastCloseAt?: number
 }
 
 export interface TrafficResponse {
   name: string
-  trafficIn: number[]
-  trafficOut: number[]
+  unit: 'bytes'
+  granularity: 'day'
+  history: TrafficPoint[]
+}
+
+export interface TrafficPoint {
+  date: string
+  trafficIn: number
+  trafficOut: number
 }

web/frps/src/utils/client.ts

@@ -10,7 +10,6 @@ export class Client {
   wireProtocol: string
   hostname: string
   ip: string
-  metas: Map<string, string>
   firstConnectedAt: Date
   lastConnectedAt: Date
   disconnectedAt?: Date
@@ -26,12 +25,6 @@ export class Client {
     this.wireProtocol = data.wireProtocol || ''
     this.hostname = data.hostname
     this.ip = data.clientIP || ''
-    this.metas = new Map<string, string>()
-    if (data.metas) {
-      for (const [key, value] of Object.entries(data.metas)) {
-        this.metas.set(key, value)
-      }
-    }
     this.firstConnectedAt = new Date(data.firstConnectedAt * 1000)
     this.lastConnectedAt = new Date(data.lastConnectedAt * 1000)
     if (data.disconnectedAt && data.disconnectedAt > 0) {
@@ -52,10 +45,6 @@ export class Client {
     return this.runID
   }
 
-  get shortRunId(): string {
-    return this.runID.substring(0, 8)
-  }
-
   get wireProtocolLabel(): string {
     if (!this.wireProtocol) return ''
     return `Protocol ${this.wireProtocol}`
@@ -73,28 +62,4 @@ export class Client {
     if (!this.disconnectedAt) return ''
     return formatDistanceToNow(this.disconnectedAt)
   }
-
-  get statusColor(): string {
-    return this.online ? 'success' : 'danger'
-  }
-
-  get metasArray(): Array<{ key: string; value: string }> {
-    const arr: Array<{ key: string; value: string }> = []
-    this.metas.forEach((value, key) => {
-      arr.push({ key, value })
-    })
-    return arr
-  }
-
-  matchesFilter(searchText: string): boolean {
-    const search = searchText.toLowerCase()
-    return (
-      this.key.toLowerCase().includes(search) ||
-      this.user.toLowerCase().includes(search) ||
-      this.clientID.toLowerCase().includes(search) ||
-      this.runID.toLowerCase().includes(search) ||
-      this.wireProtocol.toLowerCase().includes(search) ||
-      this.hostname.toLowerCase().includes(search)
-    )
-  }
 }

web/frps/src/utils/format.ts

@@ -19,6 +19,14 @@ export function formatDistanceToNow(date: Date): string {
   return Math.floor(seconds) + ' seconds ago'
 }
 
+export function formatUnixSeconds(seconds?: number): string {
+  if (seconds == null || !Number.isFinite(seconds) || seconds <= 0) return ''
+
+  const date = new Date(seconds * 1000)
+  const pad = (value: number) => value.toString().padStart(2, '0')
+  return `${pad(date.getMonth() + 1)}-${pad(date.getDate())} ${pad(date.getHours())}:${pad(date.getMinutes())}:${pad(date.getSeconds())}`
+}
+
 export function formatFileSize(bytes: number): string {
   if (!Number.isFinite(bytes) || bytes < 0) return '0 B'
   if (bytes === 0) return '0 B'