fix(node): refuse a node's claim on another inbound's client

The sync adopts each node's reported clients through SyncInbound, which resolves
a client record by email alone — and clients.email is globally unique. A node
reporting a colliding email therefore overwrote that client's UUID even when the
client is attached only to a master inbound, and the master then rebuilt its own
Xray config with the node-supplied credential: the real user locked out.

Skip a reported client whose record is attached only to inbounds of other nodes.
A record attached nowhere stays adoptable, so the soft-orphan reattach path a
flapping node depends on is unaffected.

internal/web/service/inbound_node.go

@@ -21,6 +21,8 @@ import (
 
 var reportedRemoteTagConflict sync.Map
 
+var reportedForeignClientClaim sync.Map
+
 // nodeBulkPushThreshold caps how many per-client RPCs a single operation will
 // stream to a remote node. Above it, the panel marks the node dirty instead and
 // lets one ReconcileNode push converge the whole inbound — far cheaper than M
@@ -412,6 +414,46 @@ func adoptedWireInbound(c, snapIb *model.Inbound, adoptedSettings string) *model
 	return &a
 }
 
+// clientEmailsOwnedElsewhere returns the emails attached only to inbounds of
+// other nodes: email is unique, so adopting one would overwrite a client this
+// node does not serve. Attached nowhere means soft-orphaned, hence adoptable.
+func clientEmailsOwnedElsewhere(tx *gorm.DB, nodeID int, emails []string) (map[string]struct{}, error) {
+	attachedEmails := func(nodeScoped bool) ([]string, error) {
+		q := tx.Table("clients").
+			Joins("JOIN client_inbounds ON client_inbounds.client_id = clients.id").
+			Joins("JOIN inbounds ON inbounds.id = client_inbounds.inbound_id").
+			Where("clients.email IN ?", emails)
+		if nodeScoped {
+			q = q.Where("inbounds.node_id = ?", nodeID)
+		}
+		var rows []string
+		err := q.Pluck("clients.email", &rows).Error
+		return rows, err
+	}
+	attached, err := attachedEmails(false)
+	if err != nil {
+		return nil, err
+	}
+	if len(attached) == 0 {
+		return nil, nil
+	}
+	owned, err := attachedEmails(true)
+	if err != nil {
+		return nil, err
+	}
+	ownedSet := make(map[string]struct{}, len(owned))
+	for _, email := range owned {
+		ownedSet[email] = struct{}{}
+	}
+	foreign := make(map[string]struct{})
+	for _, email := range attached {
+		if _, ok := ownedSet[email]; !ok {
+			foreign[email] = struct{}{}
+		}
+	}
+	return foreign, nil
+}
+
 func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.TrafficSnapshot, dirty, justPushed bool) (bool, error) {
 	if snap == nil || nodeID <= 0 {
 		return false, nil
@@ -1184,6 +1226,29 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi
 				}
 			}
 		}
+		if len(localEmails) > 0 {
+			foreign, err := clientEmailsOwnedElsewhere(tx, nodeID, localEmails)
+			if err != nil {
+				return false, err
+			}
+			if len(foreign) > 0 {
+				kept := filtered[:0]
+				for i := range filtered {
+					if _, claimed := foreign[filtered[i].Email]; !claimed {
+						kept = append(kept, filtered[i])
+						continue
+					}
+					key := fmt.Sprintf("%d:%s", nodeID, filtered[i].Email)
+					if _, seen := reportedForeignClientClaim.LoadOrStore(key, struct{}{}); !seen {
+						logger.Warningf(
+							"setRemoteTraffic: node %d reported client %q, which is attached only to inbounds of another node — not adopting (rename one side to remove the duplicate email)",
+							nodeID, filtered[i].Email,
+						)
+					}
+				}
+				filtered = kept
+			}
+		}
 		if err := s.clientService.SyncInbound(tx, c.Id, filtered); err != nil {
 			logger.Warningf("setRemoteTraffic: sync clients for tag %q failed: %v", snapIb.Tag, err)
 			syncFailedInbounds[c.Id] = struct{}{}