diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f3dc40a10..8a45f8287 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -80,6 +80,16 @@ jobs: exit 1 fi + - name: PostgreSQL batched stats identities + run: | + set -o pipefail + go test ./internal/web/service -run '^TestClientStatsIdentitiesPostgres$' -count=1 -v | tee /tmp/postgres-stats-identities.log + passed=$(grep -c -- '^--- PASS' /tmp/postgres-stats-identities.log || true) + if [ "$passed" -ne 1 ]; then + echo "expected 1 passing PostgreSQL stats identities test, got $passed" >&2 + exit 1 + fi + - name: PostgreSQL global-traffic heartbeat regressions run: | set -o pipefail diff --git a/docs/architecture.md b/docs/architecture.md index b9000176c..4c60361ed 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -320,6 +320,12 @@ Restart is debounced via an atomic "need restart" flag (`SetToNeedRestart` / **Key files:** `service/xray.go` (orchestration), `xray/hot_diff.go` (the diff algorithm), `xray/process.go` (process lifecycle), `xray/api.go` (gRPC calls), `xray/config.go` (config model). +Inbound-list statistics read only UUID, email and subscription ID from `clients`, joined +through `client_inbounds` in batches of up to 400 distinct inbound IDs. Full client records +remain the source for Xray and subscriptions. Missing shared-client traffic is fetched by +email, including rows owned by an inbound outside the requested list; enrichment preserves +the stored traffic, lifecycle and last-online fields. + ### 5.2 Runtime abstraction — Local vs Remote (multi-node) ⭐ most important A "node" (`model.Node`) is another 3x-ui instance this panel controls. Every state-changing diff --git a/internal/web/service/bulk_clients_test.go b/internal/web/service/bulk_clients_test.go index af6ceb048..34fba1942 100644 --- a/internal/web/service/bulk_clients_test.go +++ b/internal/web/service/bulk_clients_test.go @@ -18,7 +18,7 @@ func setupBulkDB(t *testing.T) { dbtest.InitDB(t, filepath.Join(dbDir, "x-ui.db")) } -func clientsSettings(t *testing.T, clients []model.Client) string { +func clientsSettings(t testing.TB, clients []model.Client) string { t.Helper() b, err := json.Marshal(map[string][]model.Client{"clients": clients}) if err != nil { diff --git a/internal/web/service/client_link.go b/internal/web/service/client_link.go index 78faf7268..3a744dd7b 100644 --- a/internal/web/service/client_link.go +++ b/internal/web/service/client_link.go @@ -472,6 +472,42 @@ func (s *ClientService) DetachInbound(tx *gorm.DB, inboundId int) error { return tx.Where("inbound_id = ?", inboundId).Delete(&model.ClientInbound{}).Error } +type clientStatsIdentity struct { + UUID string + Email string + SubID string +} + +func (s *ClientService) listStatsIdentitiesForInbounds(tx *gorm.DB, inboundIDs []int) (map[int][]clientStatsIdentity, error) { + if tx == nil { + tx = database.GetDB() + } + out := make(map[int][]clientStatsIdentity) + for _, batch := range chunkInts(uniqueInts(inboundIDs), sqlInChunk) { + var rows []struct { + InboundID int `gorm:"column:inbound_id"` + UUID string `gorm:"column:uuid"` + Email string `gorm:"column:email"` + SubID string `gorm:"column:sub_id"` + } + err := tx.Table("clients"). + Select("client_inbounds.inbound_id, clients.uuid, clients.email, clients.sub_id"). + Joins("JOIN client_inbounds ON client_inbounds.client_id = clients.id"). + Where("client_inbounds.inbound_id IN ?", batch). + Order("client_inbounds.inbound_id ASC, clients.id ASC"). + Find(&rows).Error + if err != nil { + return nil, err + } + for _, row := range rows { + out[row.InboundID] = append(out[row.InboundID], clientStatsIdentity{ + UUID: row.UUID, Email: row.Email, SubID: row.SubID, + }) + } + } + return out, nil +} + func (s *ClientService) ListForInbound(tx *gorm.DB, inboundId int) ([]model.Client, error) { if tx == nil { tx = database.GetDB() diff --git a/internal/web/service/inbound_client_stats_benchmark_test.go b/internal/web/service/inbound_client_stats_benchmark_test.go new file mode 100644 index 000000000..e92cfa744 --- /dev/null +++ b/internal/web/service/inbound_client_stats_benchmark_test.go @@ -0,0 +1,194 @@ +package service + +import ( + "context" + "fmt" + "strings" + "sync/atomic" + "testing" + "time" + + "gorm.io/gorm/logger" + + "github.com/mhsanaei/3x-ui/v3/internal/database" + "github.com/mhsanaei/3x-ui/v3/internal/database/model" + "github.com/mhsanaei/3x-ui/v3/internal/xray" +) + +type statsSQLMeter struct { + logger.Interface + identities atomic.Int64 + traffic atomic.Int64 + elapsed atomic.Int64 +} + +func (m *statsSQLMeter) Trace(ctx context.Context, begin time.Time, fc func() (string, int64), err error) { + sql, _ := fc() + if strings.HasPrefix(sql, "SELECT") { + if strings.Contains(sql, `FROM "clients"`) || strings.Contains(sql, "FROM `clients`") { + m.identities.Add(1) + } + if strings.Contains(sql, "client_traffics") { + m.traffic.Add(1) + } + m.elapsed.Add(int64(time.Since(begin))) + } + m.Interface.Trace(ctx, begin, fc, err) +} + +type statsPGSnapshot struct { + Calls, Rows, WalRecords, WalFpi, WalBytes, SharedBlksHit, SharedBlksRead, SharedBlksDirtied float64 + TotalExecTime float64 +} + +func snapshotStatsPG(b testing.TB) statsPGSnapshot { + b.Helper() + var snapshot statsPGSnapshot + err := database.GetDB().Raw(`SELECT COALESCE(sum(calls),0) AS calls, COALESCE(sum(rows),0) AS rows, + COALESCE(sum(total_exec_time),0) AS total_exec_time, COALESCE(sum(wal_records),0) AS wal_records, + COALESCE(sum(wal_fpi),0) AS wal_fpi, COALESCE(sum(wal_bytes),0) AS wal_bytes, + COALESCE(sum(shared_blks_hit),0) AS shared_blks_hit, COALESCE(sum(shared_blks_read),0) AS shared_blks_read, + COALESCE(sum(shared_blks_dirtied),0) AS shared_blks_dirtied + FROM public.pg_stat_statements WHERE dbid = (SELECT oid FROM pg_database WHERE datname = current_database()) + AND query ILIKE 'SELECT %' AND query LIKE '%client_inbounds%' AND query LIKE '%clients%' + AND query NOT LIKE '%pg_stat_statements%'`).Scan(&snapshot).Error + if err != nil { + b.Fatal(err) + } + return snapshot +} + +func benchmarkStatsFixture(b *testing.B, n int, shared bool) ([]*model.Inbound, int) { + b.Helper() + setupScaleDB(b) + ds := seedScaleDataset(b, n, 8) + db := database.GetDB() + if err := db.Model(&model.ClientRecord{}).Where("1 = 1").Updates(map[string]any{ + "client_host_rule_ids": "[1,2,3]", "reverse": `{"tag":"bench-reverse"}`, + "wg_private_key": "kEgYjyW0XxZX9OM1cU9e0MF+YV6PM5nT8AgVr05Ey0E=", "wg_public_key": "4qR0/Rg0kG+1AqO1TX4IvxNN82KDBeLnYeqzdIKcQSc=", + "wg_allowed_ips": `["10.0.0.2/32","fd00::2/128"]`, "wg_keep_alive": 25, + }).Error; err != nil { + b.Fatal(err) + } + attached := n + if shared { + index := 0 + for i, clients := range ds.perInbound { + for _, client := range clients { + selected := index%10 == 0 + index++ + if !selected { + continue + } + var rec model.ClientRecord + if err := db.Where("email = ?", client.Email).First(&rec).Error; err != nil { + b.Fatal(err) + } + if err := db.Create(&model.ClientInbound{ClientId: rec.Id, InboundId: ds.inboundIds[(i+1)%8]}).Error; err != nil { + b.Fatal(err) + } + attached++ + } + } + } + var inbounds []*model.Inbound + if err := db.Preload("ClientStats").Order("id ASC").Find(&inbounds).Error; err != nil { + b.Fatal(err) + } + return inbounds, attached +} + +func runStatsBenchmark(b *testing.B, listMode string) { + for _, n := range []int{1000, 10000, 20000} { + for _, shared := range []bool{false, true} { + distribution := "disjoint" + if shared { + distribution = "shared" + } + b.Run(fmt.Sprintf("%d/%s", n, distribution), func(b *testing.B) { + b.StopTimer() + base, attached := benchmarkStatsFixture(b, n, shared) + db := database.GetDB() + meter := &statsSQLMeter{Interface: db.Logger} + original := db.Logger + db.Logger = meter + defer func() { db.Logger = original }() + svc := &InboundService{} + run := func() { + var inbounds []*model.Inbound + var err error + switch listMode { + case "full": + inbounds, err = svc.GetInbounds(1) + case "slim": + inbounds, err = svc.GetInboundsSlim(1) + default: + inbounds = make([]*model.Inbound, len(base)) + for i, ib := range base { + copyInbound := *ib + copyInbound.ClientStats = append([]xray.ClientTraffic(nil), ib.ClientStats...) + inbounds[i] = ©Inbound + } + svc.enrichClientStats(db, inbounds) + } + if err != nil { + b.Fatal(err) + } + count := 0 + for _, ib := range inbounds { + count += len(ib.ClientStats) + if listMode != "slim" { + for _, st := range ib.ClientStats { + if st.UUID == "" { + b.Fatalf("missing identity for %s", st.Email) + } + } + } + } + if len(inbounds) != 8 || count != attached { + b.Fatalf("list returned inbounds=%d stats=%d, want 8/%d", len(inbounds), count, attached) + } + } + run() + meter.identities.Store(0) + meter.traffic.Store(0) + meter.elapsed.Store(0) + var before statsPGSnapshot + postgres := db.Name() == "postgres" + if postgres { + before = snapshotStatsPG(b) + } + b.ReportAllocs() + b.ResetTimer() + b.StartTimer() + for range b.N { + run() + } + b.StopTimer() + b.ReportMetric(float64(meter.identities.Load())/float64(b.N), "identity-queries/op") + b.ReportMetric(float64(meter.traffic.Load())/float64(b.N), "traffic-queries/op") + b.ReportMetric(float64(meter.elapsed.Load())/float64(b.N), "sql-ns/op") + if postgres { + after := snapshotStatsPG(b) + b.Logf("PG delta N=%d iterations=%d warmup=1 calls=%.0f rows=%.0f total_exec_time_ms=%.3f wal_records=%.0f wal_fpi=%.0f wal_bytes=%.0f shared_blks_hit=%.0f shared_blks_read=%.0f shared_blks_dirtied=%.0f", n, b.N, after.Calls-before.Calls, after.Rows-before.Rows, after.TotalExecTime-before.TotalExecTime, after.WalRecords-before.WalRecords, after.WalFpi-before.WalFpi, after.WalBytes-before.WalBytes, after.SharedBlksHit-before.SharedBlksHit, after.SharedBlksRead-before.SharedBlksRead, after.SharedBlksDirtied-before.SharedBlksDirtied) + var info struct { + StatsReset string + Dealloc int64 + } + if err := db.Raw("SELECT stats_reset::text, dealloc FROM public.pg_stat_statements_info").Scan(&info).Error; err != nil { + b.Fatal(err) + } + b.Logf("PG stats_reset=%s dealloc=%d", info.StatsReset, info.Dealloc) + } + }) + } + } +} + +func BenchmarkInboundClientStats(b *testing.B) { runStatsBenchmark(b, "") } + +func BenchmarkInboundListStats(b *testing.B) { + for _, mode := range []string{"full", "slim"} { + b.Run(mode, func(b *testing.B) { runStatsBenchmark(b, mode) }) + } +} diff --git a/internal/web/service/inbound_client_stats_postgres_test.go b/internal/web/service/inbound_client_stats_postgres_test.go new file mode 100644 index 000000000..6a241482a --- /dev/null +++ b/internal/web/service/inbound_client_stats_postgres_test.go @@ -0,0 +1,40 @@ +package service + +import ( + "os" + "testing" + + "github.com/mhsanaei/3x-ui/v3/internal/testpg" +) + +func TestClientStatsIdentitiesPostgres(t *testing.T) { + if os.Getenv("XUI_DB_TYPE") != "postgres" || os.Getenv("XUI_DB_DSN") == "" { + t.Skip("requires XUI_DB_TYPE=postgres and XUI_DB_DSN") + } + cleanup, err := testpg.IsolatePackage("stats_identities") + if err != nil { + t.Fatal(err) + } + t.Cleanup(cleanup) + for _, tc := range []struct { + name string + run func(*testing.T) + }{ + {"batch", TestClientStatsIdentitiesBatchesPreloadedInbounds}, + {"chunk_boundary", TestBackfillClientStatsChunkBoundaryAndEmpty}, + {"projection_failure", TestBackfillClientStatsProjectionFailurePreservesStats}, + {"traffic_page_failure", TestBackfillClientStatsTrafficPageFailureKeepsFreshPreload}, + {"sibling_and_identity", TestEnrichClientStatsPreservesSiblingTrafficAndClientOrder}, + {"successive_identity", TestEnrichClientStats_UsesClientsTableUUIDWhenSettingsStale}, + {"transaction", TestClientStatsIdentitiesUsesSuppliedTransaction}, + } { + t.Run(tc.name, func(t *testing.T) { + caseCleanup, err := testpg.IsolatePackage("stats_case") + if err != nil { + t.Fatal(err) + } + t.Cleanup(caseCleanup) + tc.run(t) + }) + } +} diff --git a/internal/web/service/inbound_client_stats_test.go b/internal/web/service/inbound_client_stats_test.go new file mode 100644 index 000000000..6170de9ec --- /dev/null +++ b/internal/web/service/inbound_client_stats_test.go @@ -0,0 +1,215 @@ +package service + +import ( + "errors" + "fmt" + "reflect" + "testing" + + "gorm.io/gorm" + + "github.com/mhsanaei/3x-ui/v3/internal/database" + "github.com/mhsanaei/3x-ui/v3/internal/database/model" + "github.com/mhsanaei/3x-ui/v3/internal/xray" +) + +func seedStatsIdentity(t testing.TB, inboundIDs []int, email string) model.ClientRecord { + t.Helper() + db := database.GetDB() + rec := model.ClientRecord{Email: email, UUID: "uuid-" + email, SubID: "sub-" + email, Enable: true} + if err := db.Create(&rec).Error; err != nil { + t.Fatal(err) + } + for _, id := range inboundIDs { + if err := db.Create(&model.ClientInbound{ClientId: rec.Id, InboundId: id}).Error; err != nil { + t.Fatal(err) + } + } + return rec +} + +func TestClientStatsIdentitiesBatchesPreloadedInbounds(t *testing.T) { + setupBulkDB(t) + var inbounds []*model.Inbound + for i := range 8 { + ib := mkInbound(t, 57000+i, model.VLESS, `{"clients":[]}`) + rec := seedStatsIdentity(t, []int{ib.Id}, fmt.Sprintf("batch-%d", i)) + ib.ClientStats = []xray.ClientTraffic{{Email: rec.Email, Up: 123, Down: 456, UUID: "stale", SubId: "stale"}} + inbounds = append(inbounds, ib) + } + queries := countClientTableQueries(t, "stats_batch", func() { (&InboundService{}).enrichClientStats(database.GetDB(), inbounds) }) + if queries != 1 { + t.Fatalf("clients queries = %d, want 1 for eight preloaded inbounds", queries) + } + for _, ib := range inbounds { + st := ib.ClientStats[0] + if st.UUID != "uuid-"+st.Email || st.SubId != "sub-"+st.Email || st.Up != 123 || st.Down != 456 { + t.Fatalf("enriched stats = %+v", st) + } + } +} + +func TestBackfillClientStatsChunkBoundaryAndEmpty(t *testing.T) { + setupBulkDB(t) + var inbounds []*model.Inbound + for i := range 401 { + inbounds = append(inbounds, &model.Inbound{Id: i + 1}) + } + seedStatsIdentity(t, []int{1, 401}, "boundary") + if err := database.GetDB().Create(&xray.ClientTraffic{Email: "boundary", InboundId: 999, Up: 7}).Error; err != nil { + t.Fatal(err) + } + inbounds = append(inbounds, &model.Inbound{Id: 1}, &model.Inbound{Id: 401}) + queries := countClientTableQueries(t, "stats_boundary", func() { (&InboundService{}).enrichClientStats(database.GetDB(), inbounds) }) + if queries != 2 { + t.Fatalf("clients queries = %d, want 2 for 401 distinct IDs plus duplicates", queries) + } + for _, i := range []int{0, 400} { + if len(inbounds[i].ClientStats) != 1 || inbounds[i].ClientStats[0].Up != 7 { + t.Fatalf("inbound %d stats = %+v", i, inbounds[i].ClientStats) + } + } + queries = countClientTableQueries(t, "stats_empty", func() { (&InboundService{}).backfillClientStats(database.GetDB(), nil) }) + if queries != 0 { + t.Fatalf("empty request queries = %d", queries) + } +} + +func TestBackfillClientStatsProjectionFailurePreservesStats(t *testing.T) { + setupBulkDB(t) + ib := &model.Inbound{Id: 1, ClientStats: []xray.ClientTraffic{{Email: "preloaded", UUID: "old", SubId: "old-sub", Up: 5}}} + seedStatsIdentity(t, []int{1}, "preloaded") + seedStatsIdentity(t, []int{1}, "missing") + db := database.GetDB() + if err := db.Create(&xray.ClientTraffic{Email: "missing", InboundId: 2, Up: 9}).Error; err != nil { + t.Fatal(err) + } + want := append([]xray.ClientTraffic(nil), ib.ClientStats...) + injected := errors.New("stats projection unavailable") + name := "test:stats_projection_failure" + queries := 0 + if err := db.Callback().Query().Before("gorm:query").Register(name, func(tx *gorm.DB) { + if tx.Statement.Table == "clients" { + queries++ + if queries == 2 { + tx.AddError(injected) + } + } + }); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = db.Callback().Query().Remove(name) }) + inbounds := []*model.Inbound{ib} + for id := 2; id <= 401; id++ { + inbounds = append(inbounds, &model.Inbound{Id: id}) + } + (&InboundService{}).enrichClientStats(db, inbounds) + if !reflect.DeepEqual(ib.ClientStats, want) { + t.Fatalf("projection failure mutated stats: got %+v, want %+v", ib.ClientStats, want) + } +} + +func TestBackfillClientStatsTrafficPageFailureKeepsFreshPreload(t *testing.T) { + setupBulkDB(t) + db := database.GetDB() + ib := &model.Inbound{Id: 1, ClientStats: []xray.ClientTraffic{{Email: "preloaded", UUID: "old", SubId: "old-sub", Up: 5}}} + seedStatsIdentity(t, []int{1}, "preloaded") + for i := range 401 { + email := fmt.Sprintf("page-%03d", i) + seedStatsIdentity(t, []int{1}, email) + if err := db.Create(&xray.ClientTraffic{Email: email, InboundId: 2, Up: int64(i)}).Error; err != nil { + t.Fatal(err) + } + } + pages := 0 + name := "test:stats_traffic_page_failure" + if err := db.Callback().Query().Before("gorm:query").Register(name, func(tx *gorm.DB) { + if tx.Statement.Table == "client_traffics" { + pages++ + if pages == 2 { + tx.AddError(errors.New("second traffic page unavailable")) + } + } + }); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = db.Callback().Query().Remove(name) }) + (&InboundService{}).enrichClientStats(db, []*model.Inbound{ib}) + want := []xray.ClientTraffic{{Email: "preloaded", UUID: "uuid-preloaded", SubId: "sub-preloaded", Up: 5}} + if pages != 2 || !reflect.DeepEqual(ib.ClientStats, want) { + t.Fatalf("pages=%d stats=%+v, want two pages and %+v without partial append", pages, ib.ClientStats, want) + } +} + +func TestEnrichClientStatsPreservesSiblingTrafficAndClientOrder(t *testing.T) { + setupBulkDB(t) + db := database.GetDB() + for _, email := range []string{"second", "First", "absent", "FIRST"} { + ids := []int{12} + if email == "second" { + ids = append(ids, 13) + } + seedStatsIdentity(t, ids, email) + } + rows := []xray.ClientTraffic{ + {Email: "First", InboundId: 999, Enable: true, Up: 123, Down: 456, Total: 789, ExpiryTime: 12345, LastOnline: 2468, Reset: 3}, + {Email: "second", InboundId: 999, Enable: false, Up: 10, Down: 20}, + } + if err := db.Create(&rows).Error; err != nil { + t.Fatal(err) + } + prefix := xray.ClientTraffic{Email: "unlinked-prefix", Up: 77} + inbounds := []*model.Inbound{{Id: 12, ClientStats: []xray.ClientTraffic{prefix}}, {Id: 13}, {Id: 14}} + for range 2 { + (&InboundService{}).enrichClientStats(db, inbounds) + } + wantFirst := rows[0] + wantFirst.UUID = "uuid-FIRST" + wantFirst.SubId = "sub-FIRST" + wantSecond := rows[1] + wantSecond.UUID = "uuid-second" + wantSecond.SubId = "sub-second" + if want := []xray.ClientTraffic{prefix, wantSecond, wantFirst}; !reflect.DeepEqual(inbounds[0].ClientStats, want) { + t.Fatalf("stats = %+v, want %+v", inbounds[0].ClientStats, want) + } + if !reflect.DeepEqual(inbounds[1].ClientStats, []xray.ClientTraffic{wantSecond}) || len(inbounds[2].ClientStats) != 0 { + t.Fatalf("shared/empty stats = %+v / %+v", inbounds[1].ClientStats, inbounds[2].ClientStats) + } + if err := db.Model(&model.ClientRecord{}).Where("email = ?", "FIRST").Updates(map[string]any{"uuid": "", "sub_id": ""}).Error; err != nil { + t.Fatal(err) + } + (&InboundService{}).enrichClientStats(db, inbounds) + if st := inbounds[0].ClientStats[2]; st.UUID != "" || st.SubId != "" { + t.Fatalf("empty identity failed to clear stale values: %+v", st) + } +} + +func TestClientStatsIdentitiesUsesSuppliedTransaction(t *testing.T) { + setupBulkDB(t) + db := database.GetDB() + rec := seedStatsIdentity(t, []int{1}, "transaction") + tx := db.Begin() + if tx.Error != nil { + t.Fatal(tx.Error) + } + defer tx.Rollback() + if err := tx.Model(&rec).Updates(map[string]any{"uuid": "uncommitted", "sub_id": "uncommitted-sub"}).Error; err != nil { + t.Fatal(err) + } + ib := &model.Inbound{Id: 1, ClientStats: []xray.ClientTraffic{{Email: rec.Email}}} + (&InboundService{}).enrichClientStats(tx, []*model.Inbound{ib}) + if st := ib.ClientStats[0]; st.UUID != "uncommitted" || st.SubId != "uncommitted-sub" { + t.Fatalf("transaction enrichment = %+v, want uncommitted identity", st) + } + if err := tx.Rollback().Error; err != nil { + t.Fatal(err) + } + identities, err := (&ClientService{}).listStatsIdentitiesForInbounds(nil, []int{1, 1}) + if err != nil { + t.Fatal(err) + } + want := []clientStatsIdentity{{UUID: "uuid-transaction", Email: "transaction", SubID: "sub-transaction"}} + if !reflect.DeepEqual(identities[1], want) { + t.Fatalf("fallback after rollback = %+v, want %+v", identities[1], want) + } +} diff --git a/internal/web/service/inbound_clients.go b/internal/web/service/inbound_clients.go index 96a945cc0..010440de4 100644 --- a/internal/web/service/inbound_clients.go +++ b/internal/web/service/inbound_clients.go @@ -24,48 +24,47 @@ type CopyClientsResult struct { Errors []string `json:"errors"` } -// enrichClientStats resolves each inbound's clients from the clients table -// once, fills in the UUID/SubId fields on the preloaded ClientStats, and tops -// up rows owned by a sibling inbound (shared-email mode — the row is keyed on -// email so it only preloads on its owning inbound). Reading identity from the -// clients table keeps /inbounds/list in sync with the running Xray config -// when the embedded settings JSON is stale (#6436). +// enrichClientStats refreshes identities from clients while preserving sibling-owned traffic. +// The clients table, not the embedded settings JSON, matches the running Xray config (#6436). func (s *InboundService) enrichClientStats(db *gorm.DB, inbounds []*model.Inbound) { if len(inbounds) == 0 { return } clientsByInbound := s.backfillClientStats(db, inbounds) - for i, inbound := range inbounds { - clients := clientsByInbound[i] + for _, inbound := range inbounds { + clients := clientsByInbound[inbound.Id] if len(clients) == 0 || len(inbound.ClientStats) == 0 { continue } - cMap := make(map[string]model.Client, len(clients)) + cMap := make(map[string]clientStatsIdentity, len(clients)) for _, c := range clients { cMap[strings.ToLower(c.Email)] = c } for j := range inbound.ClientStats { email := strings.ToLower(inbound.ClientStats[j].Email) if c, ok := cMap[email]; ok { - inbound.ClientStats[j].UUID = c.ID + inbound.ClientStats[j].UUID = c.UUID inbound.ClientStats[j].SubId = c.SubID } } } } -// backfillClientStats tops up each inbound's preloaded ClientStats with rows -// owned by a sibling inbound: client_traffics is keyed on email, so a client -// attached to several inbounds has one row that only preloads on the inbound -// it was created on. Returns the clients-table clients per inbound for reuse -// (not the embedded settings JSON, which can lag the live UUID — #6436). -func (s *InboundService) backfillClientStats(db *gorm.DB, inbounds []*model.Inbound) [][]model.Client { - clientsByInbound := make([][]model.Client, len(inbounds)) +// backfillClientStats loads identities once, then appends missing sibling-owned traffic. +func (s *InboundService) backfillClientStats(db *gorm.DB, inbounds []*model.Inbound) map[int][]clientStatsIdentity { + ids := make([]int, 0, len(inbounds)) + for _, inbound := range inbounds { + ids = append(ids, inbound.Id) + } + clientsByInbound, err := s.clientService.listStatsIdentitiesForInbounds(db, ids) + if err != nil { + logger.Warning("backfillClientStats:", err) + return nil + } seenByInbound := make([]map[string]struct{}, len(inbounds)) missing := make(map[string]struct{}) for i, inbound := range inbounds { - clients, _ := s.clientService.ListForInbound(db, inbound.Id) - clientsByInbound[i] = clients + clients := clientsByInbound[inbound.Id] seen := make(map[string]struct{}, len(inbound.ClientStats)) for _, st := range inbound.ClientStats { if st.Email != "" { @@ -105,7 +104,7 @@ func (s *InboundService) backfillClientStats(db *gorm.DB, inbounds []*model.Inbo byEmail[strings.ToLower(st.Email)] = st } for i, inbound := range inbounds { - for _, c := range clientsByInbound[i] { + for _, c := range clientsByInbound[inbound.Id] { if c.Email == "" { continue } diff --git a/internal/web/service/inbound_clients_stale_uuid_test.go b/internal/web/service/inbound_clients_stale_uuid_test.go index df383c686..ab8b2de6c 100644 --- a/internal/web/service/inbound_clients_stale_uuid_test.go +++ b/internal/web/service/inbound_clients_stale_uuid_test.go @@ -25,7 +25,7 @@ func TestEnrichClientStats_UsesClientsTableUUIDWhenSettingsStale(t *testing.T) { if err := db.Create(ib).Error; err != nil { t.Fatalf("create inbound: %v", err) } - rec := &model.ClientRecord{Email: "stale@e", SubID: "sub1", UUID: fresh, Enable: true} + rec := &model.ClientRecord{Email: "stale@e", SubID: "fresh-sub", UUID: fresh, Enable: true} if err := db.Create(rec).Error; err != nil { t.Fatalf("create client: %v", err) } @@ -51,7 +51,22 @@ func TestEnrichClientStats_UsesClientsTableUUIDWhenSettingsStale(t *testing.T) { if got != fresh { t.Fatalf("ClientStats.UUID = %q, want fresh %q (settings still had stale %q)", got, fresh, stale) } - if got == stale { - t.Fatalf("ClientStats.UUID still carries stale settings value %q", stale) + if gotSub := inbounds[0].ClientStats[0].SubId; gotSub != "fresh-sub" { + t.Fatalf("ClientStats.SubId = %q, want fresh-sub", gotSub) + } + for _, identity := range []struct{ uuid, sub string }{ + {"33333333-3333-3333-3333-333333333333", "next-sub"}, + {"44444444-4444-4444-4444-444444444444", ""}, + } { + if err := db.Model(rec).Updates(map[string]any{"uuid": identity.uuid, "sub_id": identity.sub}).Error; err != nil { + t.Fatal(err) + } + inbounds, err = svc.GetInbounds(1) + if err != nil { + t.Fatal(err) + } + if st := inbounds[0].ClientStats[0]; st.UUID != identity.uuid || st.SubId != identity.sub { + t.Fatalf("successive read identity = %q/%q, want %q/%q", st.UUID, st.SubId, identity.uuid, identity.sub) + } } } diff --git a/internal/web/service/inbound_runtime_depleted_filter_test.go b/internal/web/service/inbound_runtime_depleted_filter_test.go index 5b022e388..e0c288f86 100644 --- a/internal/web/service/inbound_runtime_depleted_filter_test.go +++ b/internal/web/service/inbound_runtime_depleted_filter_test.go @@ -68,6 +68,19 @@ func TestRuntimeDropsDepletedClientWhoseStatsRowPointsAtSibling(t *testing.T) { requireOnlyHealthy(t, "buildInboundForLocalRuntime", emails) }) + t.Run("generated xray config", func(t *testing.T) { + a, _ := seedDepletedOnSibling(t, model.VLESS, 23351, vless) + value, present := emittedClients(t, a.Tag) + list, ok := value.([]any) + if !present || !ok || len(list) != 1 { + t.Fatalf("generated clients = %#v, want only healthy client", value) + } + client, ok := list[0].(map[string]any) + if !ok || client["email"] != "h" || client["id"] != "11111111-1111-1111-1111-11111111111e" { + t.Fatalf("generated client = %#v, want exact healthy identity", list[0]) + } + }) + t.Run("mtproto sidecar", func(t *testing.T) { a, _ := seedDepletedOnSibling(t, model.MTProto, 23321, `{"clients":[{"email":"d","secret":"`+mtprotoTestSecretA+`","enable":true},`+ diff --git a/internal/web/service/scale_helpers_test.go b/internal/web/service/scale_helpers_test.go index dcd4eaef8..133eb68b5 100644 --- a/internal/web/service/scale_helpers_test.go +++ b/internal/web/service/scale_helpers_test.go @@ -20,11 +20,9 @@ import ( "gorm.io/gorm" ) -// setupScaleDB initializes the DB for a scale benchmark on either Postgres -// (XUI_DB_TYPE=postgres + XUI_DB_DSN) or SQLite (XUI_SCALE_TEST=1, temp file; -// XUI_SCALE_DB_PATH persists the DB for manual smoke runs), and registers -// cleanup. Skips the test when neither backend is configured. -func setupScaleDB(t *testing.T) { +// setupScaleDB opens package-isolated PostgreSQL or opt-in SQLite (XUI_SCALE_TEST); otherwise it skips. +// XUI_SCALE_DB_PATH preserves SQLite for manual smoke runs instead of temporary-file cleanup. +func setupScaleDB(t testing.TB) { t.Helper() xuilogger.InitLogger(logging.ERROR) @@ -82,11 +80,9 @@ type scaleDataset struct { perInbound [][]model.Client } -// seedScaleDataset seeds n healthy clients (future expiry, unfilled quota) -// spread across numInbounds inbounds, writing inbounds, clients, -// client_inbounds and client_traffics directly in one transaction — orders of -// magnitude faster than SyncInbound and one fsync instead of thousands. -func seedScaleDataset(t *testing.T, n, numInbounds int) scaleDataset { +// seedScaleDataset writes healthy-client fixture tables directly in one transaction. +// Skipping SyncInbound keeps per-client reconciliation out of benchmark setup. +func seedScaleDataset(t testing.TB, n, numInbounds int) scaleDataset { t.Helper() db := database.GetDB() resetScaleTables(t, db, "inbounds", "clients", "client_inbounds", "client_traffics") @@ -187,10 +183,9 @@ func sampleEmails(emails []string, k int) []string { return out } -// resetScaleTables empties the given tables between sub-sizes. Postgres uses a -// single TRUNCATE ... CASCADE; SQLite deletes per table and clears the -// autoincrement counters so ids restart like RESTART IDENTITY. -func resetScaleTables(t *testing.T, db *gorm.DB, tables ...string) { +// resetScaleTables uses PostgreSQL TRUNCATE ... CASCADE with identity restart. +// SQLite deletes table by table, then clears its autoincrement counters. +func resetScaleTables(t testing.TB, db *gorm.DB, tables ...string) { t.Helper() if config.GetDBKind() == "postgres" { stmt := "TRUNCATE TABLE " + strings.Join(tables, ", ") + " RESTART IDENTITY CASCADE"