feat(inbounds): improve multi-node online attribution (#6164)

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
 }
 

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

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

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

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.