Merge pull request #5411 from fatedier/dev

Release v0.70.0

.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
+          platforms: linux/amd64,linux/arm/v7,linux/arm64,linux/ppc64le,linux/s390x
           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
+          platforms: linux/amd64,linux/arm/v7,linux/arm64,linux/ppc64le,linux/s390x
           push: true
           tags: |
             ${{ env.TAG_FRPS }}

Makefile.cross-compiles

@@ -9,7 +9,7 @@ all: build
 build: app
 
 app:
-	@set -e; $(foreach n, $(os-archs), \
+	@$(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); \

Release.md

@@ -1,3 +1,9 @@
+## Features
+
+* Expanded the frps dashboard API v2 migration across Clients, Proxies, Server Overview, Client Detail, and Proxy Detail, covering paginated users/clients/proxies, detail data, proxy traffic history, server system info, offline proxy statistics pruning, server-side pagination, search, and proxy type filtering.
+
 ## Fixes
 
-* Fixed control-session replacement leaks when frpc reconnects through a half-open TCP multiplexed connection.
+* 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.

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.2-0.20260715094513-2780ae1d8cdb
+	github.com/fatedier/golib v0.7.0
 	github.com/google/uuid v1.6.0
 	github.com/gorilla/mux v1.8.1
 	github.com/gorilla/websocket v1.5.0
@@ -13,6 +13,7 @@ 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
@@ -52,6 +53,9 @@ 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
@@ -63,6 +67,7 @@ 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.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/golib v0.7.0 h1:tMDF9ObcwVt59VUHroJOzHQjVFPLymZVMpGm9WAVwhY=
+github.com/fatedier/golib v0.7.0/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,6 +78,14 @@ 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=
@@ -138,6 +146,8 @@ 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/nathole/discovery.go

@@ -15,24 +15,31 @@
 package nathole
 
 import (
-	"errors"
 	"fmt"
 	"net"
 	"time"
 
-	"github.com/fatedier/golib/net/stun"
+	"github.com/pion/stun/v3"
 )
 
 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
@@ -52,8 +59,9 @@ type stunResponse struct {
 
 type discoverConn struct {
 	conn *net.UDPConn
-	client    *stun.Client
+
 	localAddr   net.Addr
+	messageChan chan *Message
 }
 
 func listen(localAddr string) (*discoverConn, error) {
@@ -69,50 +77,82 @@ 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,
-		client:    client,
 		localAddr:   conn.LocalAddr(),
+		messageChan: make(chan *Message, 10),
 	}, 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
 	}
-	transaction, err := stun.NewBindingTransaction(serverAddr)
+	request, err := stun.Build(stun.TransactionID, stun.BindingRequest)
 	if err != nil {
 		return nil, err
 	}
-	if err := c.conn.SetReadDeadline(time.Now().Add(responseTimeout)); 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")
-		}
-		return nil, err
-	}
 
-	resp := &stunResponse{}
-	if response.MappedAddr != nil {
-		resp.externalAddr = response.MappedAddr.String()
+	if err = request.NewTransactionID(); err != nil {
+		return nil, err
 	}
-	if response.OtherAddr != nil {
-		resp.otherAddr = response.OtherAddr.String()
+	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
+		}
+	case <-time.After(responseTimeout):
+		return nil, fmt.Errorf("wait response from stun server timeout")
+	}
+	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 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()
 	}
 	return resp, nil
 }

pkg/nathole/discovery_test.go

@@ -1,382 +0,0 @@
-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,8 +18,10 @@ import (
 	"bytes"
 	"fmt"
 	"net"
+	"strconv"
 
 	"github.com/fatedier/golib/crypto"
+	"github.com/pion/stun/v3"
 
 	"github.com/fatedier/frp/pkg/msg"
 )
@@ -46,6 +48,20 @@ 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 {

server/control.go

@@ -17,7 +17,6 @@ package server
 import (
 	"context"
 	"fmt"
-	"net"
 	"runtime/debug"
 	"sync"
 	"sync/atomic"
@@ -41,311 +40,55 @@ 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]*controlEntry
-	registry    *registry.ClientRegistry
-	closed      bool
+	ctlsByRunID map[string]*Control
 
 	mu sync.RWMutex
 }
 
-func NewControlManager(clientRegistry *registry.ClientRegistry) *ControlManager {
+func NewControlManager() *ControlManager {
 	return &ControlManager{
-		ctlsByRunID: make(map[string]*controlEntry),
-		registry:    clientRegistry,
+		ctlsByRunID: make(map[string]*Control),
 	}
 }
 
-// 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.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
-}
-
-// 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()
+func (cm *ControlManager) Add(runID string, ctl *Control) (old *Control) {
 	cm.mu.Lock()
 	defer cm.mu.Unlock()
 
-	if cm.closed || cm.ctlsByRunID[ctl.runID] != entry || entry.ctl != ctl || entry.id != ctl.controlID {
-		return false, nil
+	var ok bool
+	old, ok = cm.ctlsByRunID[runID]
+	if ok {
+		old.Replaced(ctl)
+	}
+	cm.ctlsByRunID[runID] = ctl
+	return
 }
 
-	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()
+// we should make sure if it's the same control to prevent delete a new one
+func (cm *ControlManager) Del(runID string, ctl *Control) {
 	cm.mu.Lock()
 	defer cm.mu.Unlock()
-
-	if cm.ctlsByRunID[ctl.runID] != entry || entry.ctl != ctl || entry.id != ctl.controlID {
-		return false
+	if c, ok := cm.ctlsByRunID[runID]; ok && c == ctl {
+		delete(cm.ctlsByRunID, runID)
 	}
-	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) {
-	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")
-	}
+	defer cm.mu.RUnlock()
+	ctl, ok = cm.ctlsByRunID[runID]
+	return
 }
 
 func (cm *ControlManager) Close() error {
 	cm.mu.Lock()
-	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()
+	defer cm.mu.Unlock()
+	for _, ctl := range cm.ctlsByRunID {
+		ctl.Close()
 	}
+	cm.ctlsByRunID = make(map[string]*Control)
 	return nil
 }
 
@@ -367,20 +110,12 @@ 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
@@ -407,26 +142,16 @@ type Control struct {
 	// last time got the Ping message
 	lastPing atomic.Value
 
-	// runID never changes during the lifetime of a control. controlID is assigned
-	// once by ControlManager and distinguishes same-runID generations.
+	// 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
-	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{}
-	serverMetrics metrics.ServerMetrics
 }
 
 func NewControl(ctx context.Context, sessionCtx *SessionContext) (*Control, error) {
@@ -438,11 +163,9 @@ func NewControl(ctx context.Context, sessionCtx *SessionContext) (*Control, erro
 		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())
 
@@ -452,121 +175,48 @@ func NewControl(ctx context.Context, sessionCtx *SessionContext) (*Control, erro
 	return ctl, nil
 }
 
-func (ctl *Control) RunID() string {
-	return ctl.runID
-}
-
-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) setHandoffBarrier(barrier <-chan struct{}) {
-	ctl.lifecycleMu.Lock()
-	ctl.handoffBarrier = barrier
-	ctl.lifecycleMu.Unlock()
-}
-
-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) 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{})
 		}
-
-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()
+	ctl.sessionCtx.Conn.Close()
+	return nil
 }
 
 func (ctl *Control) Replaced(newCtl *Control) {
-	ctl.markReplaced()
-	ctl.xl.Infof("replaced by client [%s] (control ID %d)", newCtl.runID, newCtl.ID())
-	_ = ctl.interruptReadAndClose()
+	xl := ctl.xl
+	xl.Infof("replaced by client [%s]", newCtl.runID)
+	ctl.runID = ""
+	ctl.sessionCtx.Conn.Close()
 }
 
-// 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()
+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()))
+		}
+	}()
 
-	switch ctl.state {
-	case controlStateCreated:
-		ctl.state = controlStateClosing
-		ctl.finishLocked()
+	select {
+	case ctl.workConnCh <- conn:
+		xl.Debugf("new work connection registered")
 		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
+		xl.Debugf("work connection pool is full, discarding")
+		return fmt.Errorf("work connection pool is full, discarding")
 	}
 }
 
-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.
@@ -621,10 +271,10 @@ func (ctl *Control) heartbeatWorker() {
 	}
 
 	xl := ctl.xl
-	wait.Until(func() {
+	go 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.Close()
+			ctl.sessionCtx.Conn.Close()
 			return
 		}
 	}, time.Second, ctl.doneCh)
@@ -639,14 +289,14 @@ func (ctl *Control) loginUserInfo() plugin.UserInfo {
 	return plugin.UserInfo{
 		User:  ctl.sessionCtx.LoginMsg.User,
 		Metas: ctl.sessionCtx.LoginMsg.Metas,
-		RunID: ctl.runID,
+		RunID: ctl.sessionCtx.LoginMsg.RunID,
 	}
 }
 
 func (ctl *Control) closeProxy(pxy proxy.Proxy) {
 	pxy.Close()
 	ctl.sessionCtx.PxyManager.Del(pxy.GetName())
-	ctl.serverMetrics.CloseProxy(pxy.GetName(), pxy.GetConfigurer().GetBaseConfig().Type)
+	metrics.Server.CloseProxy(pxy.GetName(), pxy.GetConfigurer().GetBaseConfig().Type)
 
 	notifyContent := &plugin.CloseProxyContent{
 		User: ctl.loginUserInfo(),
@@ -661,24 +311,12 @@ 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.lifecycleMu.Lock()
-	if ctl.state == controlStateRunning {
-		ctl.state = controlStateClosing
-	}
-	ctl.lifecycleMu.Unlock()
-	_ = ctl.interruptReadAndClose()
+	ctl.sessionCtx.Conn.Close()
 
 	ctl.mu.Lock()
 	close(ctl.workConnCh)
@@ -693,14 +331,10 @@ func (ctl *Control) worker() {
 		ctl.closeProxy(pxy)
 	}
 
-	ctl.serverMetrics.CloseClient()
-	if ctl.manager != nil {
-		ctl.manager.Remove(ctl)
-	}
+	metrics.Server.CloseClient()
+	ctl.sessionCtx.ClientRegistry.MarkOfflineByRunID(ctl.runID)
 	xl.Infof("client exit success")
-	ctl.lifecycleMu.Lock()
-	ctl.finishLocked()
-	ctl.lifecycleMu.Unlock()
+	close(ctl.doneCh)
 }
 
 func (ctl *Control) registerMsgHandlers() {
@@ -740,9 +374,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.runID
+			clientID = ctl.sessionCtx.LoginMsg.RunID
 		}
-		ctl.serverMetrics.NewProxy(inMsg.ProxyName, inMsg.ProxyType, ctl.sessionCtx.LoginMsg.User, clientID)
+		metrics.Server.NewProxy(inMsg.ProxyName, inMsg.ProxyType, ctl.sessionCtx.LoginMsg.User, clientID)
 	}
 	_ = ctl.msgDispatcher.Send(resp)
 }

server/control_test.go

@@ -1,543 +0,0 @@
-// 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/registry/registry.go

@@ -28,7 +28,6 @@ type ClientInfo struct {
 	User             string
 	RawClientID      string
 	RunID            string
-	ControlID        uint64
 	Hostname         string
 	IP               string
 	Version          string
@@ -65,16 +64,6 @@ 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
 	}
@@ -94,16 +83,6 @@ func (cr *ClientRegistry) RegisterWithControlID(
 	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{
@@ -118,7 +97,6 @@ func (cr *ClientRegistry) RegisterWithControlID(
 
 	info.RawClientID = rawClientID
 	info.RunID = runID
-	info.ControlID = controlID
 	info.Hostname = hostname
 	info.IP = remoteAddr
 	info.Version = version
@@ -136,16 +114,6 @@ func (cr *ClientRegistry) RegisterWithControlID(
 
 // 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()
 
@@ -153,24 +121,18 @@ func (cr *ClientRegistry) markOfflineByRunID(runID string, controlID uint64, mat
 	if !ok {
 		return
 	}
-	if info, ok := cr.clients[key]; ok && info.RunID == runID && (!matchControlID || info.ControlID == controlID) {
+	if info, ok := cr.clients[key]; ok && info.RunID == runID {
 		if info.RawClientID == "" {
 			delete(cr.clients, key)
 		} else {
-			setClientOffline(info, cr.clock.Now())
-		}
-	}
-	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
+			now := cr.clock.Now()
 			info.DisconnectedAt = now
 		}
+	}
+	delete(cr.runIndex, runID)
+}
 
 // List returns a snapshot of all known clients.
 func (cr *ClientRegistry) List() []ClientInfo {

server/registry/registry_test.go

@@ -72,89 +72,3 @@ 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,7 +18,6 @@ import (
 	"bytes"
 	"context"
 	"crypto/tls"
-	"errors"
 	"fmt"
 	"io"
 	"net"
@@ -52,6 +51,7 @@ 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,8 +64,6 @@ 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.
@@ -163,10 +161,9 @@ func NewService(cfg *v1.ServerConfig) (*Service, error) {
 		return nil, err
 	}
 
-	clientRegistry := registry.NewClientRegistry()
 	svr := &Service{
-		ctlManager:     NewControlManager(clientRegistry),
-		clientRegistry: clientRegistry,
+		ctlManager:     NewControlManager(),
+		clientRegistry: registry.NewClientRegistry(),
 		pxyManager:     proxy.NewManager(),
 		pluginManager:  plugin.NewManager(),
 		rc: &controller.ResourceController{
@@ -472,9 +469,6 @@ 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(),
@@ -483,27 +477,29 @@ func (svr *Service) handleConnection(ctx context.Context, conn net.Conn, interna
 			}); writeErr != nil {
 				xl.Warnf("write login error response error: %v", writeErr)
 			}
-			if ctl != nil {
-				_ = ctl.Close()
-			} else {
 			conn.Close()
-			}
 			return
 		}
-		if err = svr.completeControlLogin(ctl, func() error {
-			return writeWithDeadline(conn, connWriteTimeout, func() error {
+		if err = writeWithDeadline(conn, connWriteTimeout, func() error {
 			return acceptedConn.conn.WriteMsg(&msg.LoginResp{
 				Version: version.Full(),
 				RunID:   ctl.runID,
 				Error:   "",
 			})
-			})
 		}); err != nil {
-			xl.Warnf("complete control login error: %v", err)
-			svr.ctlManager.Remove(ctl)
-			_ = ctl.Close()
+			xl.Warnf("write login response error: %v", err)
+			svr.ctlManager.Del(m.RunID, ctl)
+			svr.clientRegistry.MarkOfflineByRunID(m.RunID)
+			conn.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{
@@ -531,17 +527,6 @@ 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
@@ -791,6 +776,7 @@ func (svr *Service) RegisterControl(
 		Conn:           ctlConn,
 		LoginMsg:       loginMsg,
 		ServerCfg:      svr.cfg,
+		ClientRegistry: svr.clientRegistry,
 		WireProtocol:   wireProtocol,
 	})
 	if err != nil {
@@ -799,17 +785,18 @@ func (svr *Service) RegisterControl(
 		return nil, fmt.Errorf("unexpected error when creating new controller")
 	}
 
-	if err := svr.ctlManager.Add(ctl); err != nil {
-		return ctl, err
+	if oldCtl := svr.ctlManager.Add(loginMsg.RunID, ctl); oldCtl != nil {
+		oldCtl.WaitClosed()
 	}
-	ctl.WaitForHandoff()
 
-	active, err := svr.ctlManager.Activate(ctl)
-	if err != nil {
-		return ctl, err
+	remoteAddr := ctlConn.RemoteAddr().String()
+	if host, _, err := net.SplitHostPort(remoteAddr); err == nil {
+		remoteAddr = host
 	}
-	if !active {
-		return ctl, errControlReplaced
+	_, 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)
 	}
 
 	return ctl, nil
@@ -843,25 +830,20 @@ func (svr *Service) RegisterWorkConn(workConn *msg.Conn, newMsg *msg.NewWorkConn
 		xl.Warnf("invalid NewWorkConn with run id [%s]", newMsg.RunID)
 		return err
 	}
-	return svr.ctlManager.RegisterWorkConn(ctl, proxy.NewWorkConn(workConn))
+	return ctl.RegisterWorkConn(proxy.NewWorkConn(workConn))
 }
 
 func (svr *Service) RegisterVisitorConn(visitorConn net.Conn, newMsg *msg.NewVisitorConn, wireProtocol string) error {
-	admit := func(visitorUser string) error {
-		return svr.rc.VisitorManager.NewConn(newMsg.ProxyName, visitorConn, newMsg.Timestamp, newMsg.SignKey,
-			newMsg.UseEncryption, newMsg.UseCompression, visitorUser, wireProtocol)
-	}
+	visitorUser := ""
 	// 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 != "" {
-		admitted, err := svr.ctlManager.admitVisitorByRunID(newMsg.RunID, admit)
-		if err != nil {
-			return err
-		}
-		if !admitted {
+		ctl, exist := svr.ctlManager.GetByID(newMsg.RunID)
+		if !exist {
 			return fmt.Errorf("no client control found for run id [%s]", newMsg.RunID)
 		}
-		return nil
+		visitorUser = ctl.sessionCtx.LoginMsg.User
 	}
-	return admit("")
+	return svr.rc.VisitorManager.NewConn(newMsg.ProxyName, visitorConn, newMsg.Timestamp, newMsg.SignKey,
+		newMsg.UseEncryption, newMsg.UseCompression, visitorUser, wireProtocol)
 }

server/service_test.go

@@ -15,27 +15,12 @@
 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) {
@@ -76,504 +61,3 @@ 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,9 +94,11 @@ func (f *Framework) RunProcessesWithBinaries(
 }
 
 func (f *Framework) RunFrps(args ...string) (*process.Process, string, error) {
-	p, output, err := f.StartFrps(args...)
+	p := process.NewWithEnvs(TestContext.FRPServerPath, args, f.osEnvs)
+	f.serverProcesses = append(f.serverProcesses, p)
+	err := p.Start()
 	if err != nil {
-		return p, output, err
+		return p, p.Output(), err
 	}
 	select {
 	case <-p.Done():
@@ -105,39 +107,17 @@ 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

@@ -1,208 +0,0 @@
-// 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

@@ -1,300 +0,0 @@
-// 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)
-		}
-	}
-}