xray: let the last entry for a user win within one user batch

backend/xray/config.go

@@ -62,7 +62,22 @@ type Inbound struct {
 	clients        map[string]api.Account // Runtime-only map: email -> account (never serialized)
 }
 
+func lastEntryPerEmail(users []*common.User) []*common.User {
+	index := make(map[string]int, len(users))
+	unique := make([]*common.User, 0, len(users))
+	for _, user := range users {
+		if i, seen := index[user.GetEmail()]; seen {
+			unique[i] = user
+			continue
+		}
+		index[user.GetEmail()] = len(unique)
+		unique = append(unique, user)
+	}
+	return unique
+}
+
 func (c *Config) syncUsers(users []*common.User) {
+	users = lastEntryPerEmail(users)
 	for _, i := range c.InboundConfigs {
 		if i.exclude {
 			continue
@@ -108,7 +123,7 @@ func (c *Config) buildInboundUpdates(users []*common.User) (map[string]*Inbound,
 		}
 	}
 
-	for _, user := range users {
+	for _, user := range lastEntryPerEmail(users) {
 		settings, _ := setupUserAccount(user)
 		userInbounds := user.GetInbounds()
 		userEmail := user.GetEmail()

backend/xray/user_sync_batch_test.go

@@ -0,0 +1,80 @@
+package xray
+
+import (
+	"slices"
+	"testing"
+
+	"github.com/pasarguard/node/common"
+)
+
+func inboundEmails(cfg *Config, tag string) []string {
+	for _, inbound := range cfg.InboundConfigs {
+		if inbound.Tag != tag {
+			continue
+		}
+		inbound.mu.RLock()
+		emails := make([]string, 0, len(inbound.clients))
+		for email := range inbound.clients {
+			emails = append(emails, email)
+		}
+		inbound.mu.RUnlock()
+		slices.Sort(emails)
+		return emails
+	}
+	return nil
+}
+
+func TestUpdateUsersLetsTheLastEntryForAUserWin(t *testing.T) {
+	x, service := newRecordingXray(t)
+	ctx := syncContext(t)
+	others := []string{"vless", "vmess", "ss", "ss2022", "hysteria"}
+
+	alice := recordingUser("1.alice", "trojan")
+	rotated := recordingUser("1.alice", "trojan")
+	rotated.Proxies.Trojan.Password = "rotated trojan secret"
+
+	if err := x.UpdateUsers(ctx, []*common.User{alice}); err != nil {
+		t.Fatal(err)
+	}
+	service.take()
+
+	if err := x.UpdateUsers(ctx, []*common.User{rotated, alice}); err != nil {
+		t.Fatal(err)
+	}
+	assertOperations(t, service.take(), operations(removed(others...)...))
+
+	if err := x.UpdateUsers(ctx, []*common.User{recordingUser("1.alice"), alice}); err != nil {
+		t.Fatal(err)
+	}
+	assertOperations(t, service.take(), operations(removed(others...)...))
+	if emails := inboundEmails(x.config, "trojan"); !slices.Equal(emails, []string{"1.alice"}) {
+		t.Fatalf("the user whose last entry keeps them on trojan was dropped: trojan holds %v", emails)
+	}
+}
+
+func TestUpdateUsersAndSyncUsersLetTheLastEntryForAUserWinInTheConfig(t *testing.T) {
+	enabled := recordingUser("1.alice", "trojan")
+	disabled := recordingUser("1.alice")
+
+	for name, apply := range map[string]func(*Config, []*common.User){
+		"updateUsers": (*Config).updateUsers,
+		"syncUsers":   (*Config).syncUsers,
+	} {
+		t.Run(name, func(t *testing.T) {
+			cfg, err := NewConfig(recordingInbounds, nil)
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			apply(cfg, []*common.User{enabled, disabled})
+			if emails := inboundEmails(cfg, "trojan"); len(emails) != 0 {
+				t.Fatalf("a user whose last entry removes them is on trojan: %v", emails)
+			}
+
+			apply(cfg, []*common.User{disabled, enabled})
+			if emails := inboundEmails(cfg, "trojan"); len(emails) != 1 {
+				t.Fatalf("a user whose last entry adds them is not on trojan: %v", emails)
+			}
+		})
+	}
+}

backend/xray/user_sync_interleave_test.go

@@ -0,0 +1,72 @@
+package xray
+
+import (
+	"slices"
+	"strings"
+	"sync"
+	"testing"
+
+	"github.com/pasarguard/node/common"
+)
+
+func coreHoldsAfter(start bool, ordered []string, tag string) bool {
+	present := start
+	for _, operation := range ordered {
+		switch operation {
+		case tag + ":add":
+			present = true
+		case tag + ":remove":
+			present = false
+		}
+	}
+	return present
+}
+
+func TestConcurrentSyncUserAndUpdateUsersKeepCoreAndConfigInAgreement(t *testing.T) {
+	x, service := newRecordingXray(t)
+
+	on := recordingUser("1.alice", "trojan")
+	off := recordingUser("1.alice")
+
+	if err := x.UpdateUsers(syncContext(t), []*common.User{on}); err != nil {
+		t.Fatal(err)
+	}
+	service.takeOrdered()
+	present := true
+
+	for round := 0; round < 200; round++ {
+		ctx := syncContext(t)
+		single, batch := on, []*common.User{on, off}
+		if round%2 == 1 {
+			single, batch = off, []*common.User{off, on}
+		}
+
+		var wg sync.WaitGroup
+		wg.Add(2)
+		go func() {
+			defer wg.Done()
+			_ = x.SyncUser(ctx, single)
+		}()
+		go func() {
+			defer wg.Done()
+			_ = x.UpdateUsers(ctx, batch)
+		}()
+		wg.Wait()
+
+		ordered := service.takeOrdered()
+		present = coreHoldsAfter(present, ordered, "trojan")
+		inConfig := slices.Contains(inboundEmails(x.config, "trojan"), "1.alice")
+		if present != inConfig {
+			t.Fatalf("round %d: core holds alice on trojan = %v, config holds her = %v, operations %s", round, present, inConfig, strings.Join(ordered, " "))
+		}
+
+		if err := x.UpdateUsers(ctx, batch); err != nil {
+			t.Fatal(err)
+		}
+		present = coreHoldsAfter(present, service.takeOrdered(), "trojan")
+		want := slices.Contains(batch[len(batch)-1].GetInbounds(), "trojan")
+		if present != want || slices.Contains(inboundEmails(x.config, "trojan"), "1.alice") != want {
+			t.Fatalf("round %d: after a lone batch whose last entry wants alice on trojan = %v, core = %v", round, want, present)
+		}
+	}
+}