perf(nodes): reuse one pooled client per node instead of rebuilding it (#6548)

* perf(nodes): reuse one pooled client per node instead of rebuilding it The heartbeat probe asks for a client every 5s per node, and for skip, pin and mtls modes HTTPClientForNode built a client with its own transport each time: every tick paid a full TCP+TLS handshake per node, which is the CPU a 100-node fleet reports. Cache the client per node identity, close the previous one when that identity changes, and raise the idle pool caps above any real fleet size so a node's connection survives to its next tick. * perf(nodes): keep one client per node in the pooled cache Round-1 findings on this PR. The eviction dropped only entries whose key did not start with the current identity, so every proxy variant of that identity stayed for the life of the process. That variant is often a fresh loopback port: withOutboundBridge mints one per call and tears the bridge down on return, so each operator "test node" or remote-inbounds action added a client whose key can never be hit again, and a node switched to verify mode orphaned its old entry by returning before the loop. Replacing that filter with one entry per node bounds the cache at the fleet size, and the verify-mode return now clears the node too. TestHTTPClientForNodeKeepsOneClientPerNode fails without this -- watched red, "2, want 1" -- and pins the verify-mode cleanup on the same cache. * style(nodes): keep the eviction comment inside the two-line cap
2026-09-16 07:07:14 +00:00 · 2026-09-15 15:29:49 +03:00
parent 574caa63e9
commit 4a8fdceed6
2 changed files with 219 additions and 26 deletions
@@ -6,6 +6,7 @@ import (
 	"crypto/tls"
 	"encoding/base64"
 	"encoding/hex"
+	"fmt"
 	"net/http"
 	"strings"
 	"sync"
@@ -132,22 +133,104 @@ func (t *credentialRotatingTransport) CloseIdleConnections() {
 	current.CloseIdleConnections()
 }

-// defaultNodeHTTPClient reaches nodes trusting the system CA store ("verify"
-// mode or plain http); shared so connections pool across nodes.
-var defaultNodeHTTPClient = &http.Client{
-	Transport: &http.Transport{
-		MaxIdleConns:        64,
-		MaxIdleConnsPerHost: 4,
+// The global cap must exceed the fleet size: below it Go closes a node's
+// connection before its next heartbeat, costing a handshake every tick.
+const (
+	maxIdleNodeConns        = 512
+	maxIdleNodeConnsPerHost = 8
+)
+
+func newNodeTransport(tlsCfg *tls.Config) *http.Transport {
+	return &http.Transport{
+		MaxIdleConns:        maxIdleNodeConns,
+		MaxIdleConnsPerHost: maxIdleNodeConnsPerHost,
 		IdleConnTimeout:     60 * time.Second,
 		DialContext:         netsafe.SSRFGuardedDialContext,
-	},
+		TLSClientConfig:     tlsCfg,
+	}
 }

+// defaultNodeHTTPClient reaches nodes trusting the system CA store ("verify"
+// mode or plain http); shared so connections pool across nodes.
+var defaultNodeHTTPClient = &http.Client{Transport: newNodeTransport(nil)}
+
+// nodeClients caches one client per node: heartbeat and traffic sync reach it
+// every few seconds, and a rebuilt client would open its own empty pool.
+type nodeClientEntry struct {
+	nodeID int
+	client *http.Client
+}
+
+var (
+	nodeClientsMu    sync.Mutex
+	nodeClientsCache = map[string]nodeClientEntry{}
+)
+
+// nodeClientIdentity covers everything that decides how the node is trusted; the
+// proxy URL is a variant of it, so it stays out of the identity itself.
+func nodeClientIdentity(n *model.Node, mode string) string {
+	return fmt.Sprintf("%d|%s|%s|%s|%d|%s", n.Id, mode, n.Scheme, n.Address, n.Port, n.PinnedCertSha256)
+}
+
+// dropNodeClients discards every cached client of one node except keep, so a
+// node never holds more than the variant it is using now. Callers hold the lock.
+func dropNodeClients(nodeID int, keep string) {
+	for key, entry := range nodeClientsCache {
+		if entry.nodeID != nodeID || key == keep {
+			continue
+		}
+		entry.client.CloseIdleConnections()
+		delete(nodeClientsCache, key)
+	}
+}
+
+// HTTPClientForNode returns the pooled client for n, building it on first use
+// and whenever the node's identity or TLS material changes.
 func HTTPClientForNode(n *model.Node, proxyURL string) (*http.Client, error) {
 	mode := n.TlsVerifyMode
 	if mode == "" {
 		mode = "verify"
 	}
+	if mode == "verify" || n.Scheme == "http" {
+		// Shared across nodes and not node-specific: nothing to key on.
+		if proxyURL == "" {
+			nodeClientsMu.Lock()
+			dropNodeClients(n.Id, "")
+			nodeClientsMu.Unlock()
+			return defaultNodeHTTPClient, nil
+		}
+	}
+
+	identity := nodeClientIdentity(n, mode)
+	key := identity + "|" + proxyURL
+	nodeClientsMu.Lock()
+	if entry, ok := nodeClientsCache[key]; ok {
+		nodeClientsMu.Unlock()
+		return entry.client, nil
+	}
+	nodeClientsMu.Unlock()
+
+	client, err := buildNodeHTTPClient(n, mode, proxyURL)
+	if err != nil {
+		return nil, err
+	}
+
+	nodeClientsMu.Lock()
+	if entry, ok := nodeClientsCache[key]; ok {
+		// A concurrent caller won the race; keep its client and drop ours.
+		nodeClientsMu.Unlock()
+		client.CloseIdleConnections()
+		return entry.client, nil
+	}
+	// Any other variant is dead weight: a stale identity's pool fits no trust
+	// decision now, and an ephemeral proxy URL is never asked for twice.
+	dropNodeClients(n.Id, key)
+	nodeClientsCache[key] = nodeClientEntry{nodeID: n.Id, client: client}
+	nodeClientsMu.Unlock()
+	return client, nil
+}
+
+func buildNodeHTTPClient(n *model.Node, mode, proxyURL string) (*http.Client, error) {
 	if proxyURL != "" {
 		if mode == "mtls" && n.Scheme != "http" {
 			timeout := remoteHTTPTimeout
@@ -191,22 +274,13 @@ func HTTPClientForNode(n *model.Node, proxyURL string) (*http.Client, error) {
 		transport.TLSClientConfig = tlsCfg
 		return client, nil
 	}
-	if mode == "verify" || n.Scheme == "http" {
-		return defaultNodeHTTPClient, nil
-	}
 	if mode == "mtls" {
 		build := func() (idleClosingRoundTripper, error) {
 			tlsCfg, err := tlsConfigForNode(n)
 			if err != nil {
 				return nil, err
 			}
-			return &http.Transport{
-				MaxIdleConns:        64,
-				MaxIdleConnsPerHost: 4,
-				IdleConnTimeout:     60 * time.Second,
-				DialContext:         netsafe.SSRFGuardedDialContext,
-				TLSClientConfig:     tlsCfg,
-			}, nil
+			return newNodeTransport(tlsCfg), nil
 		}
 		transport, err := newCredentialRotatingTransport(build)
 		if err != nil {
@@ -218,15 +292,7 @@ func HTTPClientForNode(n *model.Node, proxyURL string) (*http.Client, error) {
 	if err != nil {
 		return nil, err
 	}
-	return &http.Client{
-		Transport: &http.Transport{
-			MaxIdleConns:        64,
-			MaxIdleConnsPerHost: 4,
-			IdleConnTimeout:     60 * time.Second,
-			DialContext:         netsafe.SSRFGuardedDialContext,
-			TLSClientConfig:     tlsCfg,
-		},
-	}, nil
+	return &http.Client{Transport: newNodeTransport(tlsCfg)}, nil
 }

 func tlsConfigForNode(n *model.Node) (*tls.Config, error) {
@@ -6,6 +6,8 @@ import (
 	"crypto/tls"
 	"encoding/base64"
 	"encoding/hex"
+	"io"
+	"net"
 	"net/http"
 	"net/http/httptest"
 	"net/url"
@@ -17,6 +19,7 @@ import (

 	"github.com/mhsanaei/3x-ui/v3/internal/database/model"
 	"github.com/mhsanaei/3x-ui/v3/internal/util/crypto"
+	"github.com/mhsanaei/3x-ui/v3/internal/util/netsafe"
 )

 type generationProbeTransport struct {
@@ -77,6 +80,130 @@ func TestCredentialRotatingTransportDropsOldPoolBeforeNextRequest(t *testing.T)
 	}
 }

+// Heartbeat and traffic sync ask for a client every few seconds; a rebuilt one
+// owns an empty pool, so each tick paid a fresh TCP+TLS handshake per node.
+func TestHTTPClientForNodeReusesOneConnectionAcrossCalls(t *testing.T) {
+	var handshakes atomic.Int32
+	server := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(http.StatusOK)
+	}))
+	server.Config.ConnState = func(_ net.Conn, state http.ConnState) {
+		if state == http.StateNew {
+			handshakes.Add(1)
+		}
+	}
+	server.StartTLS()
+	defer server.Close()
+
+	u, err := url.Parse(server.URL)
+	if err != nil {
+		t.Fatalf("parse test server url: %v", err)
+	}
+	port, err := strconv.Atoi(u.Port())
+	if err != nil {
+		t.Fatalf("test server port: %v", err)
+	}
+	node := &model.Node{
+		Id: 31, Address: u.Hostname(), Port: port, Scheme: "https",
+		TlsVerifyMode: "skip", AllowPrivateAddress: true,
+	}
+
+	for tick := range 2 {
+		client, err := HTTPClientForNode(node, "")
+		if err != nil {
+			t.Fatalf("tick %d: HTTPClientForNode: %v", tick, err)
+		}
+		req, err := http.NewRequestWithContext(
+			netsafe.ContextWithAllowPrivate(context.Background(), true), http.MethodGet, server.URL, nil)
+		if err != nil {
+			t.Fatalf("tick %d: new request: %v", tick, err)
+		}
+		resp, err := client.Do(req)
+		if err != nil {
+			t.Fatalf("tick %d: request: %v", tick, err)
+		}
+		_, _ = io.Copy(io.Discard, resp.Body)
+		_ = resp.Body.Close()
+	}
+
+	if got := handshakes.Load(); got != 1 {
+		t.Fatalf("TLS handshakes = %d, want 1: a rebuilt client re-handshakes on every tick", got)
+	}
+}
+
+// A node that switches to pinning (or gains a proxy) must not be served by the
+// client built for its previous trust decision.
+func TestHTTPClientForNodeRebuildsWhenNodeIdentityChanges(t *testing.T) {
+	pin := base64.StdEncoding.EncodeToString(make([]byte, sha256.Size))
+	node := &model.Node{Id: 32, Address: "node.example.test", Port: 443, Scheme: "https", TlsVerifyMode: "skip"}
+	skipped, err := HTTPClientForNode(node, "")
+	if err != nil {
+		t.Fatalf("skip client: %v", err)
+	}
+
+	pinned := *node
+	pinned.TlsVerifyMode = "pin"
+	pinned.PinnedCertSha256 = pin
+	pinnedClient, err := HTTPClientForNode(&pinned, "")
+	if err != nil {
+		t.Fatalf("pin client: %v", err)
+	}
+	if skipped == pinnedClient {
+		t.Fatal("a pinned node must not reuse the client built to skip verification")
+	}
+
+	proxied, err := HTTPClientForNode(node, "socks5://127.0.0.1:1080")
+	if err != nil {
+		t.Fatalf("proxied client: %v", err)
+	}
+	if skipped == proxied {
+		t.Fatal("a proxied node must not reuse the direct client")
+	}
+
+	if again, err := HTTPClientForNode(node, ""); err != nil || again == pinnedClient {
+		t.Fatalf("a skip request must never be served the pinned client; again=%p err=%v", again, err)
+	}
+}
+
+func nodeClientEntries(id int) int {
+	nodeClientsMu.Lock()
+	defer nodeClientsMu.Unlock()
+	count := 0
+	for _, entry := range nodeClientsCache {
+		if entry.nodeID == id {
+			count++
+		}
+	}
+	return count
+}
+
+// withOutboundBridge mints a fresh loopback port per call, so the variant it
+// asks for can never be hit again; only the variant in use may stay cached.
+func TestHTTPClientForNodeKeepsOneClientPerNode(t *testing.T) {
+	node := &model.Node{Id: 77, Address: "node.example.test", Port: 443, Scheme: "https", TlsVerifyMode: "skip"}
+	variants := []string{"socks5://127.0.0.1:41001", "socks5://127.0.0.1:41002", ""}
+	for _, variant := range variants {
+		if _, err := HTTPClientForNode(node, variant); err != nil {
+			t.Fatalf("HTTPClientForNode(%q): %v", variant, err)
+		}
+		if got := nodeClientEntries(node.Id); got != 1 {
+			t.Fatalf("cached clients for the node after %q = %d, want 1", variant, got)
+		}
+		if client, err := HTTPClientForNode(node, variant); err != nil || client == nil {
+			t.Fatalf("repeat HTTPClientForNode(%q): client=%p err=%v", variant, client, err)
+		}
+	}
+
+	verify := *node
+	verify.TlsVerifyMode = "verify"
+	if client, err := HTTPClientForNode(&verify, ""); err != nil || client != defaultNodeHTTPClient {
+		t.Fatalf("verify client = %p, want the shared one (%p); err=%v", client, defaultNodeHTTPClient, err)
+	}
+	if got := nodeClientEntries(node.Id); got != 0 {
+		t.Fatalf("cached clients for a node now on verify = %d, want 0", got)
+	}
+}
+
 func TestReloadMasterClientConnectionsValidatesProviderBeforeInvalidation(t *testing.T) {
 	before := masterCertEpoch.Load()
 	SetMasterClientCertProvider(func() (tls.Certificate, error) {