From af466b6a244d126ce85f28f8c08730279e9ccedc Mon Sep 17 00:00:00 2001 From: Sanaei Date: Tue, 15 Sep 2026 16:28:31 +0200 Subject: [PATCH] fix(node): push a node only the client IPs it hosts A master's per-node sync must scope what it sends to the clients that node serves, so its cost tracks the node and not the fleet. The global-usage push already did (node_client_traffics by node_id); the 10s client-IP push sent GetAllInboundClientIps, the whole table, to every node. Each node's MergeInboundClientIps then created a row for every foreign email, and its next GET clientIps echoed the whole fleet back. Its IP-limit job only ever reads rows for its own clients, so none of it was used. With 150 nodes x 150 clients, one IP tick pushed 299 MB and pulled 264 MB, every node held 22,500 rows instead of 150, and sync ticks grew 3.8s -> 10.2s even at 1ms latency; the cost grows with the square of the fleet. Both pushes now share nodeHostedEmails. After the change the same fleet moves 2.0 MB / 1.8 MB per tick and ticks stay near 3.2s. Nodes upgraded with foreign rows shed them within 30 minutes via pruneStaleIpRows. --- .../web/job/node_traffic_sync_ip_push_test.go | 88 +++++++++++++++++++ internal/web/job/node_traffic_sync_job.go | 2 +- internal/web/service/inbound_client_ips.go | 27 ++++++ .../web/service/inbound_traffic_global.go | 6 +- 4 files changed, 118 insertions(+), 5 deletions(-) create mode 100644 internal/web/job/node_traffic_sync_ip_push_test.go diff --git a/internal/web/job/node_traffic_sync_ip_push_test.go b/internal/web/job/node_traffic_sync_ip_push_test.go new file mode 100644 index 000000000..82ab5ac8a --- /dev/null +++ b/internal/web/job/node_traffic_sync_ip_push_test.go @@ -0,0 +1,88 @@ +package job + +import ( + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "path/filepath" + "slices" + "strconv" + "strings" + "sync" + "testing" + "time" + + "github.com/op/go-logging" + + "github.com/mhsanaei/3x-ui/v3/internal/database" + "github.com/mhsanaei/3x-ui/v3/internal/database/model" + xuilogger "github.com/mhsanaei/3x-ui/v3/internal/logger" + "github.com/mhsanaei/3x-ui/v3/internal/web/runtime" + "github.com/mhsanaei/3x-ui/v3/internal/web/service" +) + +// A node's IP-limit job only reads rows for its own clients, so pushing the +// whole table made every node store and echo back the entire fleet's IPs. +func TestNodeTrafficSyncPushesOnlyHostedClientIps(t *testing.T) { + xuilogger.InitLogger(logging.ERROR) + if err := database.InitDB(filepath.Join(t.TempDir(), "x-ui.db")); err != nil { + t.Fatalf("InitDB: %v", err) + } + t.Cleanup(func() { _ = database.CloseDB() }) + service.StartTrafficWriter() + t.Cleanup(service.StopTrafficWriter) + runtime.SetManager(runtime.NewManager(runtime.LocalDeps{APIPort: func() int { return 0 }, SetNeedRestart: func() {}})) + t.Cleanup(func() { runtime.SetManager(nil) }) + + var mu sync.Mutex + pushed := map[string][]string{} + now := time.Now().Unix() + for i, email := range []string{"a@node", "b@node"} { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + switch { + case strings.HasSuffix(r.URL.Path, "inbounds/list"): + settings := fmt.Sprintf(`{"clients":[{"email":%q,"id":"0000000%d-0000-4000-8000-000000000000","enable":true}],"decryption":"none"}`, email, i) + ib, _ := json.Marshal([]map[string]any{{ + "id": 1, "tag": fmt.Sprintf("in-%d", 20000+i), "port": 20000 + i, "protocol": "vless", "enable": true, + "settings": settings, "streamSettings": `{"network":"tcp"}`, "sniffing": `{}`, + "clientStats": []map[string]any{{"email": email, "enable": true}}, + }}) + _, _ = w.Write([]byte(`{"success":true,"obj":` + string(ib) + `}`)) + return + case strings.HasSuffix(r.URL.Path, "server/clientIps") && r.Method == http.MethodPost: + var rows []model.InboundClientIps + _ = json.NewDecoder(r.Body).Decode(&rows) + mu.Lock() + for _, row := range rows { + pushed[email] = append(pushed[email], row.ClientEmail) + } + mu.Unlock() + } + _, _ = w.Write([]byte(`{"success":true}`)) + })) + t.Cleanup(srv.Close) + host, port, _ := strings.Cut(strings.TrimPrefix(srv.URL, "http://"), ":") + portNum, _ := strconv.Atoi(port) + if err := database.GetDB().Create(&model.Node{ + Name: email, Scheme: "http", Address: host, Port: portNum, BasePath: "/", ApiToken: "tok", + Enable: true, Status: "online", AllowPrivateAddress: true, TlsVerifyMode: "verify", + }).Error; err != nil { + t.Fatalf("create node: %v", err) + } + if err := database.GetDB().Create(&model.InboundClientIps{ + ClientEmail: email, Ips: fmt.Sprintf(`[{"ip":"10.0.0.%d","timestamp":%d}]`, i+1, now), + }).Error; err != nil { + t.Fatalf("seed client ips: %v", err) + } + } + + NewNodeTrafficSyncJob().Run() + + for _, email := range []string{"a@node", "b@node"} { + if got := pushed[email]; !slices.Equal(got, []string{email}) { + t.Errorf("node hosting %s received IP rows for %v, want only [%s]", email, got, email) + } + } +} diff --git a/internal/web/job/node_traffic_sync_job.go b/internal/web/job/node_traffic_sync_job.go index f130b1283..4bc9548ab 100644 --- a/internal/web/job/node_traffic_sync_job.go +++ b/internal/web/job/node_traffic_sync_job.go @@ -447,7 +447,7 @@ func (j *NodeTrafficSyncJob) syncOne(mgr *runtime.Manager, n *model.Node, doIpSy logger.Warningf("node traffic sync: fetch client ips from %s failed: %v", n.Name, err) } - masterIps, err := j.inboundService.GetAllInboundClientIps() + masterIps, err := j.inboundService.GetNodeInboundClientIps(n.Id) if err != nil { logger.Warningf("node traffic sync: load client ips for push to %s failed: %v", n.Name, err) return active diff --git a/internal/web/service/inbound_client_ips.go b/internal/web/service/inbound_client_ips.go index 80c0d75bd..bb281deb1 100644 --- a/internal/web/service/inbound_client_ips.go +++ b/internal/web/service/inbound_client_ips.go @@ -19,6 +19,33 @@ func (s *InboundService) GetAllInboundClientIps() ([]model.InboundClientIps, err return ips, err } +// nodeHostedEmails is every client one node serves, its descendants' included. +// Per-node pushes are scoped to it so their cost tracks the node, not the fleet. +func nodeHostedEmails(db *gorm.DB, nodeID int) ([]string, error) { + var emails []string + err := db.Model(&model.NodeClientTraffic{}).Where("node_id = ?", nodeID).Pluck("email", &emails).Error + return emails, err +} + +// GetNodeInboundClientIps returns the IP rows of the clients nodeID hosts: a node's +// IP-limit job reads no other row, so pushing the rest only made it echo them back. +func (s *InboundService) GetNodeInboundClientIps(nodeID int) ([]model.InboundClientIps, error) { + db := database.GetDB() + emails, err := nodeHostedEmails(db, nodeID) + if err != nil || len(emails) == 0 { + return nil, err + } + var ips []model.InboundClientIps + for _, batch := range chunkStrings(emails, sqlInChunk) { + var page []model.InboundClientIps + if err := db.Where("client_email IN ?", batch).Find(&page).Error; err != nil { + return nil, err + } + ips = append(ips, page...) + } + return ips, nil +} + // clientIpStaleAfterSeconds mirrors job.ipStaleAfterSeconds: client IPs older than // 30 minutes are evicted. Applying the same cutoff inside the cross-node merge keeps // the synced blob bounded and stops the master's push-back from resurrecting IPs that diff --git a/internal/web/service/inbound_traffic_global.go b/internal/web/service/inbound_traffic_global.go index 6e88f73e6..2d0e98f95 100644 --- a/internal/web/service/inbound_traffic_global.go +++ b/internal/web/service/inbound_traffic_global.go @@ -168,10 +168,8 @@ func overlayGlobalTrafficValues(db *gorm.DB, rows []xray.ClientTraffic) { // its own aggregate. func (s *InboundService) GetNodeClientTraffics(nodeID int) ([]*xray.ClientTraffic, error) { db := database.GetDB() - var emails []string - if err := db.Model(&model.NodeClientTraffic{}). - Where("node_id = ?", nodeID). - Pluck("email", &emails).Error; err != nil { + emails, err := nodeHostedEmails(db, nodeID) + if err != nil { return nil, err } if len(emails) == 0 {