From a84bbeab2eb6e5b0329c6399bb66c2605839faf2 Mon Sep 17 00:00:00 2001 From: Sanaei Date: Tue, 15 Sep 2026 19:27:38 +0200 Subject: [PATCH] fix(node): drop online clients and sub-nodes of nodes no longer synced What the master derives from a node's reports (online clients, active inbounds, learned sub-nodes) must live only while that node is still synced; ClearNodeOnlineClients states it: a downed node must not keep its clients listed as online. Only a failed snapshot fetch cleared the online set, and only a failed probe cleared sub-nodes. A disabled node (both jobs skip it), a node marked offline before the sync tick reached it, a deleted node, and a node whose snapshot fetched but failed to merge all kept their clients online in onlineClients, onlineByGuid and activeInbounds, which the dashboard and a parent master's /clients/onlines read. Disabled and deleted nodes also kept their sub-nodes on the Nodes page until the panel restarted. The traffic sync now keeps online sets only for enabled, online nodes in its list, the heartbeat keeps sub-nodes only for enabled listed nodes, both before the empty-list return, and a failed merge clears like a failed fetch. The sync job's one-line call has no job-level test: that package cannot install the xray process, so RetainSyncedNodeOnlineClients carries the tested rule. --- .../job/node_heartbeat_descendants_test.go | 98 +++++++++++++++++++ internal/web/job/node_heartbeat_job.go | 1 + internal/web/job/node_traffic_sync_job.go | 1 + internal/web/service/inbound_node.go | 18 ++++ internal/web/service/node_tree.go | 16 +++ .../web/service/node_unsynced_online_test.go | 69 +++++++++++++ internal/xray/process.go | 17 ++++ 7 files changed, 220 insertions(+) create mode 100644 internal/web/job/node_heartbeat_descendants_test.go create mode 100644 internal/web/service/node_unsynced_online_test.go diff --git a/internal/web/job/node_heartbeat_descendants_test.go b/internal/web/job/node_heartbeat_descendants_test.go new file mode 100644 index 000000000..f9882dac4 --- /dev/null +++ b/internal/web/job/node_heartbeat_descendants_test.go @@ -0,0 +1,98 @@ +package job + +import ( + "net/http" + "net/http/httptest" + "path/filepath" + "strconv" + "strings" + "testing" + + "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" +) + +func transitiveGuids(t *testing.T) []string { + t.Helper() + tree, err := (&service.NodeService{}).GetNodeTree() + if err != nil { + t.Fatalf("GetNodeTree: %v", err) + } + var out []string + for _, n := range tree { + if n.Transitive { + out = append(out, n.Guid) + } + } + return out +} + +// The heartbeat skips a disabled node and never sees a deleted one, so the +// sub-nodes it had learned from them stayed on the Nodes page for good. +func TestHeartbeatDropsSubNodesOfNodesItNoLongerProbes(t *testing.T) { + cases := []struct { + name string + retire func(t *testing.T, nodeID int) + }{ + {"disabled", func(t *testing.T, nodeID int) { + if err := (&service.NodeService{}).SetEnable(nodeID, false); err != nil { + t.Fatalf("SetEnable: %v", err) + } + }}, + {"deleted", func(t *testing.T, nodeID int) { + if err := (&service.NodeService{}).Delete(nodeID); err != nil { + t.Fatalf("Delete: %v", err) + } + }}, + } + for _, tc := range cases { + t.Run(tc.name, func(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() }) + runtime.SetManager(runtime.NewManager(runtime.LocalDeps{APIPort: func() int { return 0 }, SetNeedRestart: func() {}})) + t.Cleanup(func() { runtime.SetManager(nil) }) + + 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, "server/status"): + _, _ = w.Write([]byte(`{"success":true,"obj":{"panelGuid":"direct-guid","xray":{"state":"running"}}}`)) + case strings.HasSuffix(r.URL.Path, "server/descendants"): + _, _ = w.Write([]byte(`{"success":true,"obj":[{"guid":"sub-guid","parentGuid":"direct-guid","name":"sub","status":"online"}]}`)) + default: + _, _ = w.Write([]byte(`{"success":true}`)) + } + })) + t.Cleanup(srv.Close) + host, port, _ := strings.Cut(strings.TrimPrefix(srv.URL, "http://"), ":") + portNum, _ := strconv.Atoi(port) + node := &model.Node{ + Name: "direct", Scheme: "http", Address: host, Port: portNum, BasePath: "/", ApiToken: "tok", + Enable: true, Status: "unknown", AllowPrivateAddress: true, TlsVerifyMode: "verify", + } + if err := database.GetDB().Create(node).Error; err != nil { + t.Fatalf("create node: %v", err) + } + + hb := NewNodeHeartbeatJob() + hb.Run() + if got := transitiveGuids(t); len(got) != 1 || got[0] != "sub-guid" { + t.Fatalf("sub-nodes after first heartbeat = %v, want [sub-guid]", got) + } + + tc.retire(t, node.Id) + hb.Run() + if got := transitiveGuids(t); len(got) != 0 { + t.Fatalf("sub-nodes after the node was %s = %v, want none", tc.name, got) + } + }) + } +} diff --git a/internal/web/job/node_heartbeat_job.go b/internal/web/job/node_heartbeat_job.go index bb37d7cf7..0fb1b3dde 100644 --- a/internal/web/job/node_heartbeat_job.go +++ b/internal/web/job/node_heartbeat_job.go @@ -39,6 +39,7 @@ func (j *NodeHeartbeatJob) Run() { logger.Warning("node heartbeat: load nodes failed:", err) return } + j.nodeService.RetainEnabledNodeDescendants(nodes) if len(nodes) == 0 { return } diff --git a/internal/web/job/node_traffic_sync_job.go b/internal/web/job/node_traffic_sync_job.go index 4bc9548ab..5f5c7ec03 100644 --- a/internal/web/job/node_traffic_sync_job.go +++ b/internal/web/job/node_traffic_sync_job.go @@ -94,6 +94,7 @@ func (j *NodeTrafficSyncJob) Run() { logger.Warning("node traffic sync: load nodes failed:", err) return } + j.inboundService.RetainSyncedNodeOnlineClients(nodes) if len(nodes) == 0 { return } diff --git a/internal/web/service/inbound_node.go b/internal/web/service/inbound_node.go index 06a4f2169..aac21333b 100644 --- a/internal/web/service/inbound_node.go +++ b/internal/web/service/inbound_node.go @@ -355,6 +355,10 @@ func (s *InboundService) SetRemoteTraffic(nodeID int, snap *runtime.TrafficSnaps structuralChange, inner = s.setRemoteTrafficLocked(nodeID, snap, dirty, justPushed) return inner }) + if err != nil { + // As on a failed fetch: a node whose snapshot did not merge keeps no online set. + s.ClearNodeOnlineClients(nodeID) + } return structuralChange, err } @@ -1454,6 +1458,20 @@ func (s *InboundService) ClearNodeOnlineClients(nodeID int) { } } +// RetainSyncedNodeOnlineClients keeps online clients only for nodes the traffic +// sync still fetches; a node missing from nodes was deleted. +func (s *InboundService) RetainSyncedNodeOnlineClients(nodes []*model.Node) { + process := currentXrayProcess() + if process == nil { + return + } + synced := make(map[int]bool, len(nodes)) + for _, n := range nodes { + synced[n.Id] = n.Enable && n.Status == "online" + } + process.RetainNodeOnlineClients(func(nodeID int) bool { return synced[nodeID] }) +} + // panelGuid returns this panel's stable self-identifier, used to key the local // panel's own clients in the per-node online maps (#4983). func (s *InboundService) panelGuid() string { diff --git a/internal/web/service/node_tree.go b/internal/web/service/node_tree.go index ec89ba384..ec492eef6 100644 --- a/internal/web/service/node_tree.go +++ b/internal/web/service/node_tree.go @@ -86,6 +86,22 @@ func (s *NodeService) ClearDescendants(nodeID int) { nodeDescendantsMu.Unlock() } +// RetainEnabledNodeDescendants drops sub-nodes learned from nodes the heartbeat no +// longer probes: disabled ones it skips, deleted ones missing from nodes. +func (s *NodeService) RetainEnabledNodeDescendants(nodes []*model.Node) { + enabled := make(map[int]bool, len(nodes)) + for _, n := range nodes { + enabled[n.Id] = n.Enable + } + nodeDescendantsMu.Lock() + for nodeID := range nodeDescendantsCache { + if !enabled[nodeID] { + delete(nodeDescendantsCache, nodeID) + } + } + nodeDescendantsMu.Unlock() +} + func cachedDescendants() []model.NodeSummary { nodeDescendantsMu.RLock() defer nodeDescendantsMu.RUnlock() diff --git a/internal/web/service/node_unsynced_online_test.go b/internal/web/service/node_unsynced_online_test.go new file mode 100644 index 000000000..7a272aa1b --- /dev/null +++ b/internal/web/service/node_unsynced_online_test.go @@ -0,0 +1,69 @@ +package service + +import ( + "fmt" + "reflect" + "testing" + + "github.com/mhsanaei/3x-ui/v3/internal/database" + "github.com/mhsanaei/3x-ui/v3/internal/database/model" + "github.com/mhsanaei/3x-ui/v3/internal/web/runtime" + "github.com/mhsanaei/3x-ui/v3/internal/xray" +) + +func useOnlineTestProcess(t *testing.T) *xray.Process { + t.Helper() + previousProcess, previousResult := xrayState.snapshot() + process := xray.NewTestProcess(nil, "") + xrayState.replace(process) + t.Cleanup(func() { + xrayState.mu.Lock() + xrayState.process = previousProcess + xrayState.result = previousResult + xrayState.mu.Unlock() + }) + return process +} + +// Only a failed snapshot fetch used to clear a node's online set, so a node the +// sync stopped reaching (disabled, marked offline, deleted) kept its clients online. +func TestRetainSyncedNodeOnlineClientsDropsUnsyncedNodes(t *testing.T) { + setupConflictDB(t) + process := useOnlineTestProcess(t) + svc := InboundService{} + for id := 1; id <= 4; id++ { + guid := fmt.Sprintf("g%d", id) + svc.SetNodeOnlineTree(id, map[string][]string{guid: {guid + "@x"}}) + process.SetNodeActiveInboundTree(id, map[string][]string{guid: {"in-" + guid}}) + } + + svc.RetainSyncedNodeOnlineClients([]*model.Node{ + {Id: 1, Enable: true, Status: "online"}, + {Id: 2, Enable: false, Status: "online"}, + {Id: 3, Enable: true, Status: "offline"}, + }) + + if got, want := svc.GetOnlineClientsByGuid(), map[string][]string{"g1": {"g1@x"}}; !reflect.DeepEqual(got, want) { + t.Errorf("online by guid = %v, want %v", got, want) + } + if got, want := svc.GetActiveInboundsByGuid(), map[string][]string{"g1": {"in-g1"}}; !reflect.DeepEqual(got, want) { + t.Errorf("active inbounds by guid = %v, want %v", got, want) + } +} + +func TestSetRemoteTrafficFailureClearsNodeOnlineClients(t *testing.T) { + setupConflictDB(t) + useOnlineTestProcess(t) + svc := InboundService{} + svc.SetNodeOnlineTree(7, map[string][]string{"g7": {"a@x"}}) + if err := database.GetDB().Exec("DROP TABLE inbounds").Error; err != nil { + t.Fatalf("drop inbounds: %v", err) + } + + if _, err := svc.SetRemoteTraffic(7, &runtime.TrafficSnapshot{}, false, false); err == nil { + t.Fatal("SetRemoteTraffic succeeded without an inbounds table") + } + if got := svc.GetOnlineClientsByGuid(); len(got) != 0 { + t.Errorf("online by guid after a failed merge = %v, want none", got) + } +} diff --git a/internal/xray/process.go b/internal/xray/process.go index b2e9e7ce9..ab14f30c8 100644 --- a/internal/xray/process.go +++ b/internal/xray/process.go @@ -538,6 +538,23 @@ func (p *Process) ClearNodeOnlineClients(nodeID int) { delete(p.nodeActiveInboundTrees, nodeID) } +// RetainNodeOnlineClients drops the subtree of every direct node keep rejects: nodes +// the master stopped syncing without a failed probe (disabled, offline, deleted). +func (p *Process) RetainNodeOnlineClients(keep func(nodeID int) bool) { + p.onlineMu.Lock() + defer p.onlineMu.Unlock() + for nodeID := range p.nodeOnlineTrees { + if !keep(nodeID) { + delete(p.nodeOnlineTrees, nodeID) + } + } + for nodeID := range p.nodeActiveInboundTrees { + if !keep(nodeID) { + delete(p.nodeActiveInboundTrees, nodeID) + } + } +} + // GetUptime returns the uptime of the Xray process in seconds. func (p *Process) GetUptime() uint64 { return uint64(time.Since(p.startTime).Seconds())