diff --git a/internal/web/runtime/remote.go b/internal/web/runtime/remote.go index be3221718..c8ca0596f 100644 --- a/internal/web/runtime/remote.go +++ b/internal/web/runtime/remote.go @@ -698,8 +698,13 @@ type TrafficSnapshot struct { // OnlineEmails so the master can attribute deeply nested clients to the real // node across a chain (#4983). Empty when the node is an old build without // the per-GUID endpoint — OnlineEmails is the fallback then. - OnlineTree map[string][]string - LastOnlineMap map[string]int64 + OnlineTree map[string][]string + // ActiveInboundTree is the GUID-keyed subtree of inbound tags that carried + // traffic within the node's online grace window. Empty when the node is an + // old build without the endpoint; the master then falls back to email-only + // online attribution for that node. + ActiveInboundTree map[string][]string + LastOnlineMap map[string]int64 // HostGroups carries the node's per-inbound host overrides (TLS/SNI/ // fingerprint), fetched only when the snapshot holds a not-yet-adopted tag. HostGroups []*entity.HostGroup @@ -754,6 +759,13 @@ func (r *Remote) FetchTrafficSnapshot(ctx context.Context) (*TrafficSnapshot, er _ = json.Unmarshal(envLastOnline.Obj, &snap.LastOnlineMap) } + envActiveInbounds, err := r.do(ctx, http.MethodPost, "panel/api/clients/activeInbounds", nil) + if err != nil { + logger.Debugf("remote %s active inbounds fetch failed: %v", r.node.Name, err) + } else if len(envActiveInbounds.Obj) > 0 { + _ = json.Unmarshal(envActiveInbounds.Obj, &snap.ActiveInboundTree) + } + return snap, nil } diff --git a/internal/web/service/inbound_node.go b/internal/web/service/inbound_node.go index 252d71dda..1736f100f 100644 --- a/internal/web/service/inbound_node.go +++ b/internal/web/service/inbound_node.go @@ -1115,17 +1115,21 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi // panelGuid. Remap just that entry to the node-unique key so the // clones don't merge; descendant subtrees keep their distinct GUIDs. if _, ok := tree[nodeRow.Guid]; ok { - remapped := make(map[string][]string, len(tree)) - for g, emails := range tree { - if g == nodeRow.Guid { - g = selfKey - } - remapped[g] = emails - } - tree = remapped + tree = remapGuidTreeKey(tree, nodeRow.Guid, selfKey) } } process.SetNodeOnlineTree(nodeID, tree) + + activeTree := normalizeActiveInboundTreeTags(snap.ActiveInboundTree, tagToCentral) + if guidShared && len(activeTree) > 0 { + if _, ok := activeTree[nodeRow.Guid]; ok { + activeTree = remapGuidTreeKey(activeTree, nodeRow.Guid, selfKey) + } + } + if len(activeTree) > 0 { + activeTree = filterGuidTreeKeys(activeTree, activeInboundGuidKeys(snap.Inbounds, tagToCentral, originGuidFor)) + } + process.SetNodeActiveInboundTree(nodeID, activeTree) } return structuralChange, nil @@ -1181,23 +1185,25 @@ func (s *InboundService) GetOnlineClientsByGuid() map[string][]string { } // GetActiveInboundsByGuid returns the inbound tags that carried traffic within -// the grace window for THIS panel, under its own GUID. Remote nodes don't -// report per-inbound activity, so a GUID missing from the map means "don't -// gate" for that node's inbounds. +// the grace window, keyed by the panelGuid of the node that physically hosts +// each inbound. A GUID missing from the map means "don't gate" for that node's +// inbounds (old-build node or no active-inbound signal). func (s *InboundService) GetActiveInboundsByGuid() map[string][]string { process := currentXrayProcess() if process == nil { return map[string][]string{} } + out := process.GetMergedActiveInboundTrees() active := process.GetLocalActiveInbounds() if len(active) == 0 { - return map[string][]string{} + return out } guid := s.panelGuid() if guid == "" { - return map[string][]string{} + return out } - return map[string][]string{guid: active} + out[guid] = mergeEmails(out[guid], active) + return out } func (s *InboundService) SetNodeOnlineTree(nodeID int, tree map[string][]string) { @@ -1249,6 +1255,85 @@ func mergeEmails(a, b []string) []string { return out } +func remapGuidTreeKey(tree map[string][]string, from, to string) map[string][]string { + if from == "" || to == "" || from == to { + return tree + } + remapped := make(map[string][]string, len(tree)) + for guid, values := range tree { + if guid == from { + guid = to + } + remapped[guid] = mergeEmails(remapped[guid], values) + } + return remapped +} + +func normalizeActiveInboundTreeTags(tree map[string][]string, tagToCentral map[string]*model.Inbound) map[string][]string { + if len(tree) == 0 { + return nil + } + out := make(map[string][]string, len(tree)) + for guid, tags := range tree { + if guid == "" || len(tags) == 0 { + continue + } + seen := make(map[string]struct{}, len(tags)) + for _, tag := range tags { + if tag == "" { + continue + } + if central, ok := tagToCentral[tag]; ok && central != nil && central.Tag != "" { + tag = central.Tag + } + if _, dup := seen[tag]; dup { + continue + } + seen[tag] = struct{}{} + out[guid] = append(out[guid], tag) + } + } + if len(out) == 0 { + return nil + } + return out +} + +func activeInboundGuidKeys(inbounds []*model.Inbound, tagToCentral map[string]*model.Inbound, originGuidFor func(*model.Inbound) string) map[string]struct{} { + allowed := make(map[string]struct{}) + for _, ib := range inbounds { + if ib == nil { + continue + } + if _, ok := tagToCentral[ib.Tag]; !ok { + continue + } + if guid := originGuidFor(ib); guid != "" { + allowed[guid] = struct{}{} + } + } + return allowed +} + +func filterGuidTreeKeys(tree map[string][]string, allowed map[string]struct{}) map[string][]string { + if len(tree) == 0 || len(allowed) == 0 { + return nil + } + out := make(map[string][]string, len(tree)) + for guid, values := range tree { + if _, ok := allowed[guid]; !ok { + continue + } + if len(values) > 0 { + out[guid] = values + } + } + if len(out) == 0 { + return nil + } + return out +} + func (s *InboundService) GetClientsLastOnline() (map[string]int64, error) { db := database.GetDB() var rows []xray.ClientTraffic diff --git a/internal/web/service/node_origin_guid_test.go b/internal/web/service/node_origin_guid_test.go index 672ff34a7..2748cb1b5 100644 --- a/internal/web/service/node_origin_guid_test.go +++ b/internal/web/service/node_origin_guid_test.go @@ -1,13 +1,26 @@ package service import ( + "slices" "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 assertStringSet(t *testing.T, label string, got, want []string) { + t.Helper() + g := append([]string(nil), got...) + w := append([]string(nil), want...) + slices.Sort(g) + slices.Sort(w) + if !slices.Equal(g, w) { + t.Fatalf("%s = %v, want %v", label, got, want) + } +} + // #4983: a synced inbound's OriginNodeGuid must point at the panel that // physically hosts it. A node's own local inbound (empty origin in its // snapshot) is attributed to the node's own GUID; an inbound the node forwards @@ -131,6 +144,174 @@ func TestSetRemoteTraffic_RemapsClonedNodeOwnGuidOrigin(t *testing.T) { } } +func TestSetRemoteTraffic_RemapsActiveInboundTreeAndCentralTags(t *testing.T) { + setupConflictDB(t) + db := database.GetDB() + + 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() + }) + + // Force the remote inbound to be adopted with an n1- prefix on the master: + // the active-inbound tree still arrives with the node-local tag. + if err := db.Create(&model.Inbound{ + Tag: "shared-tag", Enable: true, Port: 1000, Protocol: model.VLESS, Settings: `{"clients":[]}`, + }).Error; err != nil { + t.Fatalf("create local conflicting inbound: %v", err) + } + + // Two cloned nodes share the same panelGuid, so the node's own active tags + // must be keyed by node:1 instead of the duplicated GUID. + for _, n := range []*model.Node{ + {Id: 1, Name: "a", Address: "10.0.0.1", Port: 2053, ApiToken: "t", Guid: "dup"}, + {Id: 2, Name: "b", Address: "10.0.0.2", Port: 2053, ApiToken: "t", Guid: "dup"}, + } { + if err := db.Create(n).Error; err != nil { + t.Fatalf("create node %s: %v", n.Name, err) + } + } + + snap := &runtime.TrafficSnapshot{ + Inbounds: []*model.Inbound{{ + Tag: "shared-tag", + Enable: true, + Port: 8443, + Protocol: model.VLESS, + Settings: `{"clients":[]}`, + }}, + ActiveInboundTree: map[string][]string{ + "dup": {"shared-tag"}, + }, + } + + svc := InboundService{} + if _, err := svc.setRemoteTrafficLocked(1, snap, false); err != nil { + t.Fatalf("setRemoteTrafficLocked: %v", err) + } + + merged := process.GetMergedActiveInboundTrees() + assertStringSet(t, "active node:1", merged["node:1"], []string{"n1-shared-tag"}) + if _, ok := merged["dup"]; ok { + t.Fatalf("cloned active-inbound subtree must not stay under shared GUID: %v", merged) + } +} + +func TestSetRemoteTraffic_NormalizesForwardedActiveInboundSubtreeTags(t *testing.T) { + setupConflictDB(t) + db := database.GetDB() + + 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() + }) + + for _, tag := range []string{"own-tag", "child-tag"} { + if err := db.Create(&model.Inbound{ + Tag: tag, Enable: true, Port: 1000, Protocol: model.VLESS, Settings: `{"clients":[]}`, + }).Error; err != nil { + t.Fatalf("create local conflicting inbound %q: %v", tag, err) + } + } + if err := db.Create(&model.Node{ + Id: 1, Name: "node2", Address: "10.0.0.2", Port: 2053, ApiToken: "t", Guid: "node2-guid", + }).Error; err != nil { + t.Fatalf("create node: %v", err) + } + + snap := &runtime.TrafficSnapshot{ + Inbounds: []*model.Inbound{ + { + Tag: "own-tag", + Enable: true, + Port: 8443, + Protocol: model.VLESS, + Settings: `{"clients":[]}`, + }, + { + Tag: "child-tag", + Enable: true, + Port: 9443, + Protocol: model.VLESS, + Settings: `{"clients":[]}`, + OriginNodeGuid: "child-guid", + }, + }, + ActiveInboundTree: map[string][]string{ + "node2-guid": {"own-tag"}, + "child-guid": {"child-tag"}, + }, + } + + svc := InboundService{} + if _, err := svc.setRemoteTrafficLocked(1, snap, false); err != nil { + t.Fatalf("setRemoteTrafficLocked: %v", err) + } + + merged := process.GetMergedActiveInboundTrees() + assertStringSet(t, "direct node active tags", merged["node2-guid"], []string{"n1-own-tag"}) + assertStringSet(t, "forwarded child active tags", merged["child-guid"], []string{"n1-child-tag"}) +} + +func TestSetRemoteTraffic_DropsForeignActiveInboundGuid(t *testing.T) { + setupConflictDB(t) + db := database.GetDB() + + 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() + }) + + for _, n := range []*model.Node{ + {Id: 1, Name: "node-a", Address: "10.0.0.1", Port: 2053, ApiToken: "t", Guid: "node-a-guid"}, + {Id: 2, Name: "node-b", Address: "10.0.0.2", Port: 2053, ApiToken: "t", Guid: "node-b-guid"}, + } { + if err := db.Create(n).Error; err != nil { + t.Fatalf("create node %s: %v", n.Name, err) + } + } + + snap := &runtime.TrafficSnapshot{ + Inbounds: []*model.Inbound{{ + Tag: "own-tag", + Enable: true, + Port: 8443, + Protocol: model.VLESS, + Settings: `{"clients":[]}`, + }}, + ActiveInboundTree: map[string][]string{ + "node-a-guid": {"own-tag"}, + "node-b-guid": {"foreign-tag"}, + }, + } + + svc := InboundService{} + if _, err := svc.setRemoteTrafficLocked(1, snap, false); err != nil { + t.Fatalf("setRemoteTrafficLocked: %v", err) + } + + merged := process.GetMergedActiveInboundTrees() + assertStringSet(t, "own active tags", merged["node-a-guid"], []string{"own-tag"}) + if _, ok := merged["node-b-guid"]; ok { + t.Fatalf("foreign active-inbound subtree should be ignored: %v", merged) + } +} + // A node mid-restart can return an empty inbound list with success=true. The // sync must NOT treat that as "delete all my inbounds" — otherwise a blip wipes // the node's central inbounds and every client on them (what happened to the diff --git a/internal/xray/online_test.go b/internal/xray/online_test.go index 0c5a3ed4e..a621d6c2f 100644 --- a/internal/xray/online_test.go +++ b/internal/xray/online_test.go @@ -118,16 +118,55 @@ func TestGetLocalActiveInboundsTracksGraceWindow(t *testing.T) { } } +func TestMergedActiveInboundTreesScopesPerGuid(t *testing.T) { + p := newOnlineTestProcess() + p.SetNodeActiveInboundTree(1, map[string][]string{ + "guid-a": {"in-a", "in-a"}, + "guid-b": {"in-b"}, + }) + p.SetNodeActiveInboundTree(2, map[string][]string{ + "guid-a": {"in-c"}, + "guid-c": {}, + }) + + merged := p.GetMergedActiveInboundTrees() + assertSameSet(t, "guid-a", merged["guid-a"], []string{"in-a", "in-c"}) + assertSameSet(t, "guid-b", merged["guid-b"], []string{"in-b"}) + if _, ok := merged["guid-c"]; ok { + t.Errorf("empty active-inbound GUID set should be omitted: %v", merged) + } +} + // TestClearNodeOnlineClientsDropsNode mirrors a failed node probe: the node's // whole subtree contribution disappears immediately. func TestClearNodeOnlineClientsDropsNode(t *testing.T) { p := newOnlineTestProcess() p.SetNodeOnlineTree(3, map[string][]string{"guid-a": {"user1"}}) + p.SetNodeActiveInboundTree(3, map[string][]string{"guid-a": {"in-a"}}) p.ClearNodeOnlineClients(3) if _, ok := p.GetMergedNodeTrees()["guid-a"]; ok { t.Errorf("node 3's subtree should be absent after ClearNodeOnlineClients") } + if _, ok := p.GetMergedActiveInboundTrees()["guid-a"]; ok { + t.Errorf("node 3's active-inbound subtree should be absent after ClearNodeOnlineClients") + } +} + +func TestSetNodeTreesEmptyInputDropsNode(t *testing.T) { + p := newOnlineTestProcess() + p.SetNodeOnlineTree(3, map[string][]string{"guid-a": {"user1"}}) + p.SetNodeActiveInboundTree(3, map[string][]string{"guid-a": {"in-a"}}) + + p.SetNodeOnlineTree(3, nil) + p.SetNodeActiveInboundTree(3, nil) + + if _, ok := p.GetMergedNodeTrees()["guid-a"]; ok { + t.Errorf("empty online tree should remove node 3's subtree") + } + if _, ok := p.GetMergedActiveInboundTrees()["guid-a"]; ok { + t.Errorf("empty active-inbound tree should remove node 3's subtree") + } } // TestOnlineAPISupportTriState pins the lazy capability probe contract: a new diff --git a/internal/xray/process.go b/internal/xray/process.go index 693d395ef..b2e9e7ce9 100644 --- a/internal/xray/process.go +++ b/internal/xray/process.go @@ -176,7 +176,12 @@ type process struct { // mutex guards this map, onlineClients, and localLastOnline above so the // online getters never see a torn read. nodeOnlineTrees map[int]map[string][]string - onlineMu sync.RWMutex + // nodeActiveInboundTrees mirrors nodeOnlineTrees for active inbound tags: + // each direct node reports a GUID-keyed subtree of inbound tags that carried + // traffic within its own grace window. The inbounds page combines this with + // nodeOnlineTrees so a multi-inbound client is not shown on an idle inbound. + nodeActiveInboundTrees map[int]map[string][]string + onlineMu sync.RWMutex // onlineAPISupport caches whether the running core implements the // online-stats RPCs (GetUsersStats). A new process is created on every @@ -398,10 +403,9 @@ func (p *Process) GetMergedNodeTrees() map[string][]string { } // GetLocalActiveInbounds returns a copy of THIS panel's inbound tags that -// carried traffic within the grace window. Only the local xray reports -// per-inbound activity; remote-node snapshots don't carry it, so the service -// layer keys these under the panel's own GUID and a node missing from the -// active-inbounds map means "don't gate" (fall back to the email-only signal). +// carried traffic within the grace window. The service layer keys these under +// the panel's own GUID before merging them with remote-node active-inbound +// subtrees. func (p *Process) GetLocalActiveInbounds() []string { p.onlineMu.RLock() defer p.onlineMu.RUnlock() @@ -413,6 +417,43 @@ func (p *Process) GetLocalActiveInbounds() []string { return out } +// GetMergedActiveInboundTrees returns the union of every direct node's reported +// active-inbound subtree, keyed by the panelGuid of the node that physically +// hosts each inbound. Duplicate tags reported through multiple paths are +// deduped per GUID. +func (p *Process) GetMergedActiveInboundTrees() map[string][]string { + p.onlineMu.RLock() + defer p.onlineMu.RUnlock() + if len(p.nodeActiveInboundTrees) == 0 { + return map[string][]string{} + } + out := make(map[string][]string) + seen := make(map[string]map[string]struct{}) + for _, tree := range p.nodeActiveInboundTrees { + for guid, tags := range tree { + if guid == "" || len(tags) == 0 { + continue + } + dedup := seen[guid] + if dedup == nil { + dedup = make(map[string]struct{}, len(tags)) + seen[guid] = dedup + } + for _, tag := range tags { + if tag == "" { + continue + } + if _, ok := dedup[tag]; ok { + continue + } + dedup[tag] = struct{}{} + out[guid] = append(out[guid], tag) + } + } + } + return out +} + // RefreshLocalOnline records that each email in activeEmails and each tag in // activeInboundTags had local xray traffic at now, then rebuilds onlineClients // and localActiveInbounds from every entry seen within graceMs, pruning older @@ -462,12 +503,31 @@ func (p *Process) RefreshLocalOnline(activeEmails, activeInboundTags []string, n func (p *Process) SetNodeOnlineTree(nodeID int, tree map[string][]string) { p.onlineMu.Lock() defer p.onlineMu.Unlock() + if len(tree) == 0 { + delete(p.nodeOnlineTrees, nodeID) + return + } if p.nodeOnlineTrees == nil { p.nodeOnlineTrees = map[int]map[string][]string{} } p.nodeOnlineTrees[nodeID] = tree } +// SetNodeActiveInboundTree records the GUID-keyed active-inbound subtree one +// direct remote node reported. Replaces any previous entry for that node. +func (p *Process) SetNodeActiveInboundTree(nodeID int, tree map[string][]string) { + p.onlineMu.Lock() + defer p.onlineMu.Unlock() + if len(tree) == 0 { + delete(p.nodeActiveInboundTrees, nodeID) + return + } + if p.nodeActiveInboundTrees == nil { + p.nodeActiveInboundTrees = map[int]map[string][]string{} + } + p.nodeActiveInboundTrees[nodeID] = tree +} + // ClearNodeOnlineClients drops a direct node's whole subtree contribution. // Called when a probe fails so a downed node — and everything behind it — doesn't // keep its clients listed as "online" until the next successful probe. @@ -475,6 +535,7 @@ func (p *Process) ClearNodeOnlineClients(nodeID int) { p.onlineMu.Lock() defer p.onlineMu.Unlock() delete(p.nodeOnlineTrees, nodeID) + delete(p.nodeActiveInboundTrees, nodeID) } // GetUptime returns the uptime of the Xray process in seconds.