mirror of
https://github.com/MHSanaei/3x-ui.git
synced 2026-09-17 07:37:15 +00:00
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.
This commit is contained in:
@@ -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)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -39,6 +39,7 @@ func (j *NodeHeartbeatJob) Run() {
|
|||||||
logger.Warning("node heartbeat: load nodes failed:", err)
|
logger.Warning("node heartbeat: load nodes failed:", err)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
j.nodeService.RetainEnabledNodeDescendants(nodes)
|
||||||
if len(nodes) == 0 {
|
if len(nodes) == 0 {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -94,6 +94,7 @@ func (j *NodeTrafficSyncJob) Run() {
|
|||||||
logger.Warning("node traffic sync: load nodes failed:", err)
|
logger.Warning("node traffic sync: load nodes failed:", err)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
j.inboundService.RetainSyncedNodeOnlineClients(nodes)
|
||||||
if len(nodes) == 0 {
|
if len(nodes) == 0 {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -355,6 +355,10 @@ func (s *InboundService) SetRemoteTraffic(nodeID int, snap *runtime.TrafficSnaps
|
|||||||
structuralChange, inner = s.setRemoteTrafficLocked(nodeID, snap, dirty, justPushed)
|
structuralChange, inner = s.setRemoteTrafficLocked(nodeID, snap, dirty, justPushed)
|
||||||
return inner
|
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
|
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
|
// panelGuid returns this panel's stable self-identifier, used to key the local
|
||||||
// panel's own clients in the per-node online maps (#4983).
|
// panel's own clients in the per-node online maps (#4983).
|
||||||
func (s *InboundService) panelGuid() string {
|
func (s *InboundService) panelGuid() string {
|
||||||
|
|||||||
@@ -86,6 +86,22 @@ func (s *NodeService) ClearDescendants(nodeID int) {
|
|||||||
nodeDescendantsMu.Unlock()
|
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 {
|
func cachedDescendants() []model.NodeSummary {
|
||||||
nodeDescendantsMu.RLock()
|
nodeDescendantsMu.RLock()
|
||||||
defer nodeDescendantsMu.RUnlock()
|
defer nodeDescendantsMu.RUnlock()
|
||||||
|
|||||||
@@ -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)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -538,6 +538,23 @@ func (p *Process) ClearNodeOnlineClients(nodeID int) {
|
|||||||
delete(p.nodeActiveInboundTrees, nodeID)
|
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.
|
// GetUptime returns the uptime of the Xray process in seconds.
|
||||||
func (p *Process) GetUptime() uint64 {
|
func (p *Process) GetUptime() uint64 {
|
||||||
return uint64(time.Since(p.startTime).Seconds())
|
return uint64(time.Since(p.startTime).Seconds())
|
||||||
|
|||||||
Reference in New Issue
Block a user