diff --git a/backend/xray/config.go b/backend/xray/config.go index 4f5bbb19..411b625d 100644 --- a/backend/xray/config.go +++ b/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() diff --git a/backend/xray/user_sync_batch_test.go b/backend/xray/user_sync_batch_test.go new file mode 100644 index 00000000..da1b86d8 --- /dev/null +++ b/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) + } + }) + } +} diff --git a/backend/xray/user_sync_interleave_test.go b/backend/xray/user_sync_interleave_test.go new file mode 100644 index 00000000..70edb498 --- /dev/null +++ b/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) + } + } +}