diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 341064b44..f3dc40a10 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -80,6 +80,16 @@ jobs: exit 1 fi + - name: PostgreSQL global-traffic heartbeat regressions + run: | + set -o pipefail + go test ./internal/web/service -run '^TestGlobalTrafficHeartbeatPostgres$' -count=1 -v | tee /tmp/postgres-global-traffic-heartbeat.log + passed=$(grep -c -- '^--- PASS' /tmp/postgres-global-traffic-heartbeat.log || true) + if [ "$passed" -ne 1 ]; then + echo "expected 1 passing global-traffic heartbeat test, got $passed" >&2 + exit 1 + fi + - name: PostgreSQL deferred-commit test run: | set -o pipefail diff --git a/docs/architecture.md b/docs/architecture.md index b2e1e959c..b9000176c 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -386,6 +386,17 @@ A client reset is queued per hosting node in `model.NodePendingReset` (`service/ and replayed by the node sync until the node accepts it. Periodic resets: `job/periodic_traffic_reset_job.go` (keyed off `Inbound.TrafficReset`). +Cross-panel totals are stored per master GUID and hosted email, separately from local +`client_traffics`. Directional counter changes, including lower values and resets, persist +on the next push. Identical totals refresh only when the persisted timestamp is at least +one hour old; pushes still send the same batches. Display and quota reads retain the +inclusive 24-hour persisted-row freshness window. Full-list snapshots remain local-only, +while slim/detail reads and downstream forwarding retain their global overlays. + +The heartbeat coarsens liveness, not counter propagation: an idle master can expire up to +one hour earlier relative to its final unchanged push. Active hourly refresh leaves roughly +23 hours of missed-push margin. No cache, external service or configurable interval is added. + ### 5.4 Background jobs (cron) All registered in `web.go` → `startTask()`. Each is a struct with a `Run()` method in `internal/web/job/`: diff --git a/internal/database/model/client_global_traffic.go b/internal/database/model/client_global_traffic.go index b260f79e0..322968e81 100644 --- a/internal/database/model/client_global_traffic.go +++ b/internal/database/model/client_global_traffic.go @@ -1,15 +1,7 @@ package model -// ClientGlobalTraffic mirrors a master panel's aggregated (global) usage for a -// client hosted on this panel. Masters push one row per (master, email) so the -// node can display the client's true cross-panel total and enforce its quota -// locally. The values never feed back into client_traffics — that table keeps -// this panel's local-only counters, which is what keeps every master's -// delta-baseline accounting over our snapshot correct. -// -// Rows are overwritten in place on every push (not max-merged), so a traffic -// reset on the master propagates here within one push cycle. Readers that need -// a single number fold the per-master rows with MAX. +// ClientGlobalTraffic separates each master's aggregate from local client_traffics for delta accounting; readers fold with MAX. +// Changed totals (including resets) overwrite on the next push; UpdatedAt tracks changes or hourly unchanged heartbeats. type ClientGlobalTraffic struct { Id int `json:"id" gorm:"primaryKey;autoIncrement"` MasterGuid string `json:"masterGuid" gorm:"uniqueIndex:idx_master_email,priority:1;not null"` diff --git a/internal/nodee2e/node_sync_test.go b/internal/nodee2e/node_sync_test.go index 410b13ea1..22fa26a29 100644 --- a/internal/nodee2e/node_sync_test.go +++ b/internal/nodee2e/node_sync_test.go @@ -342,3 +342,119 @@ func runNodeSyncScenarios(t *testing.T, bin, scope string) { } }) } + +func panelTraffic(t *testing.T, p *panel, email string) (int64, int64) { + t.Helper() + var tr struct{ Up, Down int64 } + if err := json.Unmarshal(p.call(http.MethodGet, "/panel/api/clients/traffic/"+email, nil), &tr); err != nil { + t.Fatal(err) + } + return tr.Up, tr.Down +} + +func enrollTrafficNode(t *testing.T, master, node *panel) { + t.Helper() + token := node.mintToken("traffic-upstream", "node-sync") + master.call(http.MethodPost, "/panel/api/nodes/add", map[string]any{ + "name": node.name, "scheme": "http", "address": "127.0.0.1", "port": node.port, "basePath": "/", + "apiToken": token, "enable": true, "allowPrivateAddress": true, + }) +} + +func TestNodeGlobalTrafficHeartbeatChain(t *testing.T) { + bin := panelBinary(t) + root := newPanel(t, bin, "traffic-root") + middle := newPanel(t, bin, "traffic-middle") + leaf := newPanel(t, bin, "traffic-leaf") + const email = "chain-usage" + localIDs := make(map[*panel]int) + makeClient := func() map[string]any { + client := vlessClient(email) + client["subId"] = "chain-sub" + return client + } + for _, p := range []*panel{root, middle, leaf} { + p.start() + var ib inboundView + obj := p.call(http.MethodPost, "/panel/api/inbounds/add", vlessInbound(p.name, freePort(t), nil, makeClient())) + if err := json.Unmarshal(obj, &ib); err != nil || ib.Id == 0 { + t.Fatalf("%s local inbound=%s err=%v", p.name, obj, err) + } + localIDs[p] = ib.Id + } + enrollTrafficNode(t, middle, leaf) + enrollTrafficNode(t, root, middle) + eventually(t, 90*time.Second, "chain adopts all zero-traffic baselines", func() (bool, string) { + rootCount, middleCount := len(root.inbounds()), len(middle.inbounds()) + return rootCount == 3 && middleCount == 2, fmt.Sprintf("root inbounds=%d middle inbounds=%d", rootCount, middleCount) + }) + waitUsage := func(p *panel, up, down int64) { + t.Helper() + eventually(t, 90*time.Second, p.name+" converges global traffic", func() (bool, string) { + gotUp, gotDown := panelTraffic(t, p, email) + return gotUp == up && gotDown == down, fmt.Sprintf("%s usage=%d/%d want %d/%d", p.name, gotUp, gotDown, up, down) + }) + } + root.call(http.MethodPost, "/panel/api/clients/updateTraffic/"+email, map[string]any{"upload": 100, "download": 200}) + middle.call(http.MethodPost, "/panel/api/clients/updateTraffic/"+email, map[string]any{"upload": 10, "download": 20}) + leaf.call(http.MethodPost, "/panel/api/clients/updateTraffic/"+email, map[string]any{"upload": 1, "download": 2}) + for _, p := range []*panel{root, middle, leaf} { + waitUsage(p, 111, 222) + } + pushRoot := func(up, down int64) { + root.call(http.MethodPost, "/panel/api/inbounds/pushClientTraffics", map[string]any{ + "masterGuid": "synthetic-upstream", "traffics": []map[string]any{{"email": email, "up": up, "down": down}}, + }) + } + pushRoot(900, 800) + waitUsage(middle, 900, 800) + waitUsage(leaf, 900, 800) + pushRoot(500, 400) + waitUsage(middle, 500, 400) + waitUsage(leaf, 500, 400) + // Each panel independently produced local usage; resetting an overlay cannot erase it. + // Reset those producers through their real API before testing forwarded zero counters. + for _, p := range []*panel{leaf, middle, root} { + p.call(http.MethodPost, "/panel/api/clients/resetTraffic/"+email, nil) + } + pushRoot(0, 0) + for _, p := range []*panel{root, middle, leaf} { + waitUsage(p, 0, 0) + } + leaf.stop() + pushRoot(77, 88) + waitUsage(middle, 77, 88) + leaf.start() + waitUsage(leaf, 77, 88) + for _, p := range []*panel{leaf, middle, root} { + p.call(http.MethodPost, "/panel/api/clients/del/"+email, nil) + } + for _, p := range []*panel{root, middle, leaf} { + eventually(t, 90*time.Second, p.name+" deletes chain client", func() (bool, string) { + for _, ib := range p.inbounds() { + if ib.client(email) != nil { + return false, fmt.Sprintf("%s retains %s", p.name, email) + } + } + return true, "" + }) + } + for _, p := range []*panel{leaf, middle, root} { + p.call(http.MethodPost, "/panel/api/clients/add", map[string]any{ + "client": makeClient(), "inboundIds": []int{localIDs[p]}, + }) + } + for _, p := range []*panel{root, middle, leaf} { + eventually(t, 90*time.Second, p.name+" recreates chain client", func() (bool, string) { + for _, ib := range p.inbounds() { + if ib.client(email) != nil { + return true, "" + } + } + return false, "missing recreated client" + }) + } + pushRoot(9, 8) + waitUsage(middle, 9, 8) + waitUsage(leaf, 9, 8) +} diff --git a/internal/web/service/client_paging.go b/internal/web/service/client_paging.go index d8196d192..e4a28333a 100644 --- a/internal/web/service/client_paging.go +++ b/internal/web/service/client_paging.go @@ -185,7 +185,7 @@ func newClientQuery(db *gorm.DB, nowMs, expireDiffMs, trafficDiffBytes int64) cl joins: []clientQueryJoin{{sql: "LEFT JOIN client_traffics ct ON ct.email = c.email"}}, usedExpr: "(COALESCE(ct.up, 0) + COALESCE(ct.down, 0))", } - freshSince := globalTrafficFreshSince() + freshSince := globalTrafficFreshSince(db.NowFunc()) var probe int64 err := db.Model(&model.ClientGlobalTraffic{}). Where("updated_at >= ?", freshSince). diff --git a/internal/web/service/global_traffic_heartbeat_test.go b/internal/web/service/global_traffic_heartbeat_test.go new file mode 100644 index 000000000..ce82a3118 --- /dev/null +++ b/internal/web/service/global_traffic_heartbeat_test.go @@ -0,0 +1,374 @@ +package service + +import ( + "encoding/json" + "fmt" + "sync" + "sync/atomic" + "testing" + "time" + + "gorm.io/gorm" + + "github.com/mhsanaei/3x-ui/v3/internal/database/model" + "github.com/mhsanaei/3x-ui/v3/internal/xray" +) + +func globalTrafficClock(t *testing.T, db *gorm.DB, queued bool) *atomic.Int64 { + t.Helper() + StopTrafficWriter() + old := db.NowFunc + clock := &atomic.Int64{} + clock.Store(time.Date(2026, 1, 2, 0, 0, 0, 0, time.UTC).UnixMilli()) + db.NowFunc = func() time.Time { return time.UnixMilli(clock.Load()) } + t.Cleanup(func() { + StopTrafficWriter() + db.NowFunc = old + }) + if queued { + StartTrafficWriter() + } + return clock +} + +func countGlobalTrafficWrites(t *testing.T, db *gorm.DB) *atomic.Int64 { + t.Helper() + rows := &atomic.Int64{} + const name = "test:global_traffic_affected_rows" + if err := db.Callback().Create().After("gorm:create").Register(name, func(tx *gorm.DB) { + if tx.Error == nil && tx.Statement.Table == "client_global_traffics" { + rows.Add(tx.RowsAffected) + } + }); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + if err := db.Callback().Create().Remove(name); err != nil { + t.Error(err) + } + }) + return rows +} + +func pushGlobalTraffic(t *testing.T, guid string, rows ...*xray.ClientTraffic) { + t.Helper() + if err := (&InboundService{}).AcceptGlobalTraffic(guid, rows); err != nil { + t.Fatal(err) + } +} + +func globalTrafficRow(t *testing.T, db *gorm.DB, guid, email string) model.ClientGlobalTraffic { + t.Helper() + var row model.ClientGlobalTraffic + if err := db.Where("master_guid = ? AND email = ?", guid, email).First(&row).Error; err != nil { + t.Fatal(err) + } + return row +} + +func TestGlobalTrafficHeartbeatBounds(t *testing.T) { + db := initGlobalHeartbeatDB(t) + clock := globalTrafficClock(t, db, true) + seedClientRow(t, "bounds", 1, 10, 20, 0) + writes := countGlobalTrafficWrites(t, db) + start := clock.Load() + push := func(up, down, wantWrites int64) { + t.Helper() + writes.Store(0) + pushGlobalTraffic(t, "master", &xray.ClientTraffic{Email: "bounds", Up: up, Down: down}) + if got := writes.Load(); got != wantWrites { + t.Fatalf("at %d counters %d/%d: affected rows=%d want %d", clock.Load()-start, up, down, got, wantWrites) + } + got := globalTrafficRow(t, db, "master", "bounds") + if got.Up != up || got.Down != down { + t.Fatalf("counters=%d/%d want %d/%d", got.Up, got.Down, up, down) + } + } + push(100, 200, 1) + clock.Store(start + time.Hour.Milliseconds() - 1) + push(100, 200, 0) + if got := globalTrafficRow(t, db, "master", "bounds").UpdatedAt; got != start { + t.Fatalf("early heartbeat timestamp=%d want %d", got, start) + } + clock.Add(1) + push(100, 200, 1) + if got := globalTrafficRow(t, db, "master", "bounds").UpdatedAt; got != clock.Load() { + t.Fatalf("boundary timestamp=%d want %d", got, clock.Load()) + } + clock.Add(1) + push(100, 200, 0) + for _, pair := range [][2]int64{{101, 200}, {99, 200}, {99, 201}, {99, 199}, {0, 199}, {0, 0}} { + push(pair[0], pair[1], 1) + } + local := readTraffic(t, db, "bounds") + if local.Up != 10 || local.Down != 20 { + t.Fatalf("local traffic changed: %+v", local) + } +} + +func TestGlobalTrafficHeartbeatNullableAndBatch(t *testing.T) { + db := initGlobalHeartbeatDB(t) + clock := globalTrafficClock(t, db, true) + var payload []*xray.ClientTraffic + for i := range 405 { + email := fmt.Sprintf("batch-%03d", i) + seedClientRow(t, email, 1, 0, 0, 0) + payload = append(payload, &xray.ClientTraffic{Email: email, Up: 10, Down: 20}) + } + pushGlobalTraffic(t, "master", payload[:404]...) + clock.Add(time.Minute.Milliseconds()) + for i, column := range []string{"up", "down", "updated_at"} { + if err := db.Model(&model.ClientGlobalTraffic{}).Where("email = ?", payload[i].Email).UpdateColumn(column, nil).Error; err != nil { + t.Fatal(err) + } + } + if err := db.Model(&model.ClientGlobalTraffic{}).Where("email = ?", payload[3].Email).UpdateColumn("updated_at", clock.Load()-time.Hour.Milliseconds()).Error; err != nil { + t.Fatal(err) + } + payload[4].Up = 11 + payload = append(payload, nil, &xray.ClientTraffic{Email: "ghost", Up: 999}, &xray.ClientTraffic{Email: payload[5].Email, Up: 12, Down: 23}) + writes := countGlobalTrafficWrites(t, db) + pushGlobalTraffic(t, "master", payload...) + if got := writes.Load(); got != 7 { + t.Fatalf("mixed two-full-batch payload affected %d rows want 7", got) + } + for i := range 6 { + got := globalTrafficRow(t, db, "master", payload[i].Email) + up, down := payload[i].Up, payload[i].Down + if i == 5 { + up, down = 12, 23 + } + if got.Up != up || got.Down != down || got.UpdatedAt != clock.Load() { + t.Fatalf("row %d=%+v want %d/%d at %d", i, got, up, down, clock.Load()) + } + } + var count int64 + if err := db.Model(&model.ClientGlobalTraffic{}).Count(&count).Error; err != nil || count != 405 { + t.Fatalf("known rows=%d want 405 err=%v", count, err) + } +} + +func TestGlobalTrafficHeartbeatLongIdleAndRecovery(t *testing.T) { + db := initGlobalHeartbeatDB(t) + clock := globalTrafficClock(t, db, true) + seedClientRow(t, "active", 1, 1, 2, 100) + seedClientRow(t, "absent", 1, 1, 2, 100) + ib := &model.Inbound{UserId: 1, Tag: "idle", Port: 43124, Enable: true, Protocol: model.VLESS, Settings: `{"clients":[]}`} + if err := db.Create(ib).Error; err != nil { + t.Fatal(err) + } + if err := (&ClientService{}).SyncInbound(db, ib.Id, []model.Client{ + {Email: "active", Enable: true, TotalGB: 100}, {Email: "absent", Enable: true, TotalGB: 100}, + }); err != nil { + t.Fatal(err) + } + start := clock.Load() + pushGlobalTraffic(t, "active-master", &xray.ClientTraffic{Email: "active", Up: 80, Down: 70}) + pushGlobalTraffic(t, "absent-master", &xray.ClientTraffic{Email: "absent", Up: 80, Down: 70}) + writes := countGlobalTrafficWrites(t, db) + for tick := 1; tick <= 5760; tick++ { + clock.Store(start + int64(tick)*30_000) + pushGlobalTraffic(t, "active-master", &xray.ClientTraffic{Email: "active", Up: 80, Down: 70}) + } + if got := writes.Load(); got != 48 { + t.Fatalf("48h idle pushes affected %d rows want 48", got) + } + assertUsage := func(email string, up, down int64, depleted bool) { + t.Helper() + rows := []*xray.ClientTraffic{{Email: email, Up: 1, Down: 2}} + overlayGlobalTraffic(db, rows) + if rows[0].Up != up || rows[0].Down != down { + t.Fatalf("%s overlay=%d/%d want %d/%d", email, rows[0].Up, rows[0].Down, up, down) + } + cond, args := depletedCond(db) + var count int64 + if err := db.Model(&xray.ClientTraffic{}).Where("email = ?", email).Where(cond, args...).Count(&count).Error; err != nil { + t.Fatal(err) + } + if (count == 1) != depleted { + t.Fatalf("%s depleted count=%d want depleted=%v", email, count, depleted) + } + page, err := (&ClientService{}).ListPaged(&InboundService{}, &SettingService{}, ClientPageParams{PageSize: 10, Search: email, Filter: "depleted"}) + if err != nil { + t.Fatal(err) + } + want := 0 + if depleted { + want = 1 + } + if len(page.Items) != want || (want == 1 && page.Items[0].Email != email) { + t.Fatalf("%s depleted page=%+v want %d rows", email, page.Items, want) + } + } + assertUsage("active", 80, 70, true) + assertUsage("absent", 1, 2, false) + last := clock.Load() + clock.Add(23 * time.Hour.Milliseconds()) + assertUsage("active", 80, 70, true) + clock.Store(last + 24*time.Hour.Milliseconds()) + assertUsage("active", 80, 70, true) + clock.Add(1) + assertUsage("active", 1, 2, false) + pushGlobalTraffic(t, "active-master", &xray.ClientTraffic{Email: "active", Up: 80, Down: 70}) + assertUsage("active", 80, 70, true) +} + +func TestGlobalTrafficHeartbeatConcurrentPairs(t *testing.T) { + for _, queued := range []bool{true, false} { + t.Run(fmt.Sprintf("queued=%v", queued), func(t *testing.T) { + db := initGlobalHeartbeatDB(t) + globalTrafficClock(t, db, queued) + seedClientRow(t, "concurrent", 1, 1, 2, 0) + var wg sync.WaitGroup + errs := make(chan error, 24) + start := make(chan struct{}) + for i := range 24 { + wg.Go(func() { + <-start + guid := "same" + if i >= 12 { + guid = fmt.Sprintf("master-%d", i) + } + errs <- (&InboundService{}).AcceptGlobalTraffic(guid, []*xray.ClientTraffic{{Email: "concurrent", Up: int64(i + 100), Down: int64(1000 - i)}}) + }) + } + close(start) + wg.Wait() + close(errs) + for err := range errs { + if err != nil { + t.Fatal(err) + } + } + row := globalTrafficRow(t, db, "same", "concurrent") + if row.Up < 100 || row.Up > 111 || row.Up+row.Down != 1100 { + t.Fatalf("torn submitted pair: %+v", row) + } + for i := 12; i < 24; i++ { + got := globalTrafficRow(t, db, fmt.Sprintf("master-%d", i), "concurrent") + if got.Up != int64(i+100) || got.Down != int64(1000-i) { + t.Fatalf("master %d pair=%+v", i, got) + } + } + pushGlobalTraffic(t, "same", &xray.ClientTraffic{Email: "concurrent", Up: 3, Down: 4}) + row = globalTrafficRow(t, db, "same", "concurrent") + if row.Up != 3 || row.Down != 4 { + t.Fatalf("ordered lower reset=%+v", row) + } + rows := []*xray.ClientTraffic{{Email: "concurrent", Up: 1, Down: 2}} + overlayGlobalTraffic(db, rows) + if rows[0].Up != 123 || rows[0].Down != 988 { + t.Fatalf("componentwise max=%d/%d want 123/988", rows[0].Up, rows[0].Down) + } + }) + } +} + +func TestGlobalTrafficHeartbeatQuotaAndLifecycle(t *testing.T) { + db := initGlobalHeartbeatDB(t) + globalTrafficClock(t, db, true) + svc := &InboundService{} + ib := &model.Inbound{ + UserId: 1, Tag: "quota", Port: 43123, Enable: true, Protocol: model.VLESS, + Settings: `{"clients":[{"email":"quota","enable":true,"totalGB":1000}]}`, + } + if err := db.Create(ib).Error; err != nil { + t.Fatal(err) + } + clients := &ClientService{} + if err := clients.SyncInbound(db, ib.Id, []model.Client{{Email: "quota", Enable: true, TotalGB: 1000}}); err != nil { + t.Fatal(err) + } + seedClientRow(t, "quota", ib.Id, 10, 20, 1000) + invalidations := &atomic.Int64{} + release := RegisterSubDataInvalidator(func() { invalidations.Add(1) }) + defer release() + pushGlobalTraffic(t, "master", &xray.ClientTraffic{Email: "quota", Up: 800, Down: 700}) + pushGlobalTraffic(t, "master", &xray.ClientTraffic{Email: "quota", Up: 800, Down: 700}) + if invalidations.Load() != 0 { + t.Fatalf("ingestion invalidated subscriptions %d times", invalidations.Load()) + } + if err := db.Transaction(func(tx *gorm.DB) error { + _, count, _, err := svc.disableInvalidClients(tx, newTrafficMutationBatch()) + if count != 1 { + t.Errorf("disabled rows=%d want 1", count) + } + return err + }); err != nil { + t.Fatal(err) + } + if invalidations.Load() != 1 { + t.Fatalf("quota disable invalidations=%d want 1", invalidations.Load()) + } + if err := db.First(ib, ib.Id).Error; err != nil { + t.Fatal(err) + } + var settings struct { + Clients []struct { + Email string `json:"email"` + Enable bool `json:"enable"` + } `json:"clients"` + } + if err := json.Unmarshal([]byte(ib.Settings), &settings); err != nil { + t.Fatal(err) + } + if len(settings.Clients) != 1 || settings.Clients[0].Email != "quota" || settings.Clients[0].Enable || readTraffic(t, db, "quota").Enable { + t.Fatalf("disabled settings=%s traffic=%+v", ib.Settings, readTraffic(t, db, "quota")) + } + if err := svc.ResetClientTrafficByEmail("quota"); err != nil { + t.Fatal(err) + } + assertRows := func(want int64) { + t.Helper() + var count int64 + if err := db.Model(&model.ClientGlobalTraffic{}).Where("email = ?", "quota").Count(&count).Error; err != nil || count != want { + t.Fatalf("side rows=%d want %d err=%v", count, want, err) + } + } + assertRows(0) + pushGlobalTraffic(t, "master", &xray.ClientTraffic{Email: "quota", Up: 0, Down: 0}) + assertRows(1) + if _, err := clients.DeleteByEmail(svc, "quota", false); err != nil { + t.Fatal(err) + } + assertRows(0) + pushGlobalTraffic(t, "master", &xray.ClientTraffic{Email: "quota", Up: 7, Down: 8}) + assertRows(0) + if _, err := clients.CreateOne(svc, ib.Id, model.Client{Email: "quota", Enable: true}); err != nil { + t.Fatal(err) + } + pushGlobalTraffic(t, "master", &xray.ClientTraffic{Email: "quota", Up: 7, Down: 8}) + assertRows(1) + got := globalTrafficRow(t, db, "master", "quota") + if got.Up != 7 || got.Down != 8 { + t.Fatalf("recreated counters=%+v", got) + } +} + +func TestGlobalTrafficHeartbeatForwarding(t *testing.T) { + db := initGlobalHeartbeatDB(t) + clock := globalTrafficClock(t, db, true) + seedClientRow(t, "forward", 1, 10, 20, 0) + if err := db.Create(&model.NodeClientTraffic{NodeId: 7, Email: "forward"}).Error; err != nil { + t.Fatal(err) + } + assertForward := func(up, down int64) { + t.Helper() + rows, err := (&InboundService{}).GetNodeClientTraffics(7) + if err != nil || len(rows) != 1 || rows[0].Up != up || rows[0].Down != down { + t.Fatalf("forward=%+v want %d/%d err=%v", rows, up, down, err) + } + } + pushGlobalTraffic(t, "upstream", &xray.ClientTraffic{Email: "forward", Up: 900, Down: 800}) + writes := countGlobalTrafficWrites(t, db) + clock.Add(time.Minute.Milliseconds()) + pushGlobalTraffic(t, "upstream", &xray.ClientTraffic{Email: "forward", Up: 900, Down: 800}) + assertForward(900, 800) + if writes.Load() != 0 { + t.Fatalf("unchanged forward push affected %d rows", writes.Load()) + } + pushGlobalTraffic(t, "upstream", &xray.ClientTraffic{Email: "forward", Up: 50, Down: 60}) + assertForward(50, 60) + pushGlobalTraffic(t, "upstream", &xray.ClientTraffic{Email: "forward", Up: 0, Down: 0}) + assertForward(10, 20) +} diff --git a/internal/web/service/global_traffic_postgres_test.go b/internal/web/service/global_traffic_postgres_test.go new file mode 100644 index 000000000..455643cac --- /dev/null +++ b/internal/web/service/global_traffic_postgres_test.go @@ -0,0 +1,39 @@ +package service + +import ( + "os" + "testing" + + "gorm.io/gorm" + + "github.com/mhsanaei/3x-ui/v3/internal/testpg" +) + +func initGlobalHeartbeatDB(t *testing.T) *gorm.DB { + t.Helper() + cleanup, err := testpg.IsolatePackage("global_heartbeat") + if err != nil { + t.Fatal(err) + } + t.Cleanup(cleanup) + return initTrafficTestDB(t) +} + +func TestGlobalTrafficHeartbeatPostgres(t *testing.T) { + if os.Getenv("XUI_DB_TYPE") != "postgres" || os.Getenv("XUI_DB_DSN") == "" { + t.Skip("set XUI_DB_TYPE=postgres and XUI_DB_DSN") + } + for _, tc := range []struct { + name string + run func(*testing.T) + }{ + {"bounds", TestGlobalTrafficHeartbeatBounds}, + {"quota and lifecycle", TestGlobalTrafficHeartbeatQuotaAndLifecycle}, + {"forwarding", TestGlobalTrafficHeartbeatForwarding}, + {"nullable and batches", TestGlobalTrafficHeartbeatNullableAndBatch}, + {"long idle and recovery", TestGlobalTrafficHeartbeatLongIdleAndRecovery}, + {"concurrent pairs", TestGlobalTrafficHeartbeatConcurrentPairs}, + } { + t.Run(tc.name, tc.run) + } +} diff --git a/internal/web/service/global_traffic_scale_test.go b/internal/web/service/global_traffic_scale_test.go new file mode 100644 index 000000000..fc106b673 --- /dev/null +++ b/internal/web/service/global_traffic_scale_test.go @@ -0,0 +1,118 @@ +package service + +import ( + "fmt" + "os" + "testing" + "time" + + "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" +) + +type globalTrafficPGStats struct { + Calls, Rows, WalRecords, WalFpi int64 + TotalExecTime, WalBytes float64 + SharedBlksHit, SharedBlksRead, SharedBlksDirtied int64 +} + +func globalTrafficStats(t *testing.T, db *gorm.DB) globalTrafficPGStats { + t.Helper() + var stats globalTrafficPGStats + if db.Name() != "postgres" { + return stats + } + if err := db.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)::float8 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 'INSERT INTO "client_global_traffics"%'`).Scan(&stats).Error; err != nil { + t.Fatal(err) + } + var info struct { + StatsReset time.Time + Dealloc int64 + } + if err := db.Raw("SELECT stats_reset, dealloc FROM public.pg_stat_statements_info").Scan(&info).Error; err != nil { + t.Fatal(err) + } + t.Logf("pg_stat_statements_info stats_reset=%s dealloc=%d cumulative=%+v", info.StatsReset.Format(time.RFC3339Nano), info.Dealloc, stats) + return stats +} + +func TestAcceptGlobalTrafficScale(t *testing.T) { + if os.Getenv("XUI_SCALE_TEST") != "1" { + t.Skip("set XUI_SCALE_TEST=1") + } + setupScaleDB(t) + db := database.GetDB() + for _, workload := range []string{"unchanged", "sparse", "all", "reset"} { + t.Run(workload, func(t *testing.T) { + resetScaleTables(t, db, "client_global_traffics", "client_traffics") + clock := globalTrafficClock(t, db, true) + const clients, ticks = 6708, 240 + locals := make([]xray.ClientTraffic, clients) + payload := make([]*xray.ClientTraffic, clients) + for i := range clients { + email := fmt.Sprintf("heartbeat-scale-%05d", i) + locals[i] = xray.ClientTraffic{Email: email, InboundId: 1, Enable: true} + payload[i] = &xray.ClientTraffic{Email: email, Up: 1000, Down: 2000} + } + if err := db.CreateInBatches(&locals, 200).Error; err != nil { + t.Fatal(err) + } + pushGlobalTraffic(t, "scale-master", payload...) + before := globalTrafficStats(t, db) + writes := countGlobalTrafficWrites(t, db) + startClock := clock.Load() + start := time.Now() + for tick := 1; tick <= ticks; tick++ { + clock.Store(startClock + int64(tick)*30_000) + for i, row := range payload { + switch workload { + case "sparse": + if (i+tick)%100 == 0 { + row.Up++ + row.Down++ + } + case "all": + row.Up++ + row.Down++ + case "reset": + if tick == 120 { + row.Up, row.Down = 0, 0 + } + } + } + pushGlobalTraffic(t, "scale-master", payload...) + } + elapsed := time.Since(start) + after := globalTrafficStats(t, db) + delta := globalTrafficPGStats{ + Calls: after.Calls - before.Calls, Rows: after.Rows - before.Rows, + TotalExecTime: after.TotalExecTime - before.TotalExecTime, + WalRecords: after.WalRecords - before.WalRecords, WalFpi: after.WalFpi - before.WalFpi, WalBytes: after.WalBytes - before.WalBytes, + SharedBlksHit: after.SharedBlksHit - before.SharedBlksHit, SharedBlksRead: after.SharedBlksRead - before.SharedBlksRead, + SharedBlksDirtied: after.SharedBlksDirtied - before.SharedBlksDirtied, + } + t.Logf("dialect=%s clients=%d ticks=%d warmup=one initial push workload=%s elapsed=%s affected_rows=%d pg_delta=%+v", db.Name(), clients, ticks, workload, elapsed, writes.Load(), delta) + var rows []model.ClientGlobalTraffic + if err := db.Where("master_guid = ?", "scale-master").Order("email").Find(&rows).Error; err != nil { + t.Fatal(err) + } + if len(rows) != clients { + t.Fatalf("rows=%d want %d", len(rows), clients) + } + for i, got := range rows { + if got.Up != payload[i].Up || got.Down != payload[i].Down { + t.Fatalf("client %d counters=%d/%d want %d/%d", i, got.Up, got.Down, payload[i].Up, payload[i].Down) + } + } + }) + } +} diff --git a/internal/web/service/inbound_disable.go b/internal/web/service/inbound_disable.go index 8e1c7fb98..72264a422 100644 --- a/internal/web/service/inbound_disable.go +++ b/internal/web/service/inbound_disable.go @@ -34,8 +34,8 @@ func (s *InboundService) disableInvalidInbounds(tx *gorm.DB, mutationBatch *traf const globalTrafficFreshWindow = 24 * time.Hour -func globalTrafficFreshSince() int64 { - return time.Now().Add(-globalTrafficFreshWindow).UnixMilli() +func globalTrafficFreshSince(now time.Time) int64 { + return now.Add(-globalTrafficFreshWindow).UnixMilli() } // depletedClientsCond matches clients that exhausted their quota or expired. @@ -60,21 +60,19 @@ const depletedClientsCond = `((total > 0 AND up + down >= total) const depletedClientsCondLocal = `((total > 0 AND up + down >= total) OR (expiry_time > 0 AND expiry_time <= ?))` -// depletedCond returns the predicate matching depleted clients together with -// the arguments it binds. The local-only variant is used unless this panel -// holds a global-traffic row a master still refreshes, in which case the -// cross-panel EXISTS check is needed to enforce combined quota. +// depletedCond includes cross-panel quota usage only while a master's row is fresh. +// Without fresh global rows, the local-only predicate avoids a correlated EXISTS scan. func depletedCond(tx *gorm.DB) (string, []any) { - now := time.Now().UnixMilli() - freshSince := globalTrafficFreshSince() + now := tx.NowFunc() + freshSince := globalTrafficFreshSince(now) var probe int64 err := tx.Model(&model.ClientGlobalTraffic{}). Where("updated_at >= ?", freshSince). Limit(1).Count(&probe).Error if err == nil && probe > 0 { - return depletedClientsCond, []any{now, freshSince} + return depletedClientsCond, []any{now.UnixMilli(), freshSince} } - return depletedClientsCondLocal, []any{now} + return depletedClientsCondLocal, []any{now.UnixMilli()} } func (s *InboundService) disableInvalidClients(tx *gorm.DB, mutationBatch *trafficMutationBatch) (bool, int64, []int, error) { diff --git a/internal/web/service/inbound_traffic_global.go b/internal/web/service/inbound_traffic_global.go index 2d0e98f95..46b9e3dea 100644 --- a/internal/web/service/inbound_traffic_global.go +++ b/internal/web/service/inbound_traffic_global.go @@ -13,14 +13,10 @@ import ( "gorm.io/gorm/clause" ) -// AcceptGlobalTraffic ingests a master panel's aggregated per-client usage -// into client_global_traffics, keyed by (masterGuid, email). The numbers are -// display/enforcement inputs only — client_traffics keeps this panel's -// local-only counters, so the pushing master's (and any other master's) -// delta accounting over our snapshot stays correct. -// -// Rows are overwritten, not max-merged: a reset on the master propagates here -// on its next push. Emails this panel doesn't host are dropped. +const globalTrafficHeartbeatInterval = time.Hour + +// AcceptGlobalTraffic overwrites hosted clients' per-master usage, including lower resets. +// Identical usage refreshes hourly; client_traffics remains local-only for delta accounting. func (s *InboundService) AcceptGlobalTraffic(masterGuid string, traffics []*xray.ClientTraffic) error { masterGuid = strings.TrimSpace(masterGuid) if masterGuid == "" { @@ -57,7 +53,9 @@ func (s *InboundService) AcceptGlobalTraffic(masterGuid string, traffics []*xray return nil } - now := time.Now().UnixMilli() + now := db.NowFunc() + updatedAt := now.UnixMilli() + heartbeatCutoff := now.Add(-globalTrafficHeartbeatInterval).UnixMilli() rows := make([]model.ClientGlobalTraffic, 0, len(known)) for _, email := range known { t := byEmail[email] @@ -69,7 +67,7 @@ func (s *InboundService) AcceptGlobalTraffic(masterGuid string, traffics []*xray Email: email, Up: t.Up, Down: t.Down, - UpdatedAt: now, + UpdatedAt: updatedAt, }) } @@ -78,6 +76,15 @@ func (s *InboundService) AcceptGlobalTraffic(masterGuid string, traffics []*xray if err := tx.Clauses(clause.OnConflict{ Columns: []clause.Column{{Name: "master_guid"}, {Name: "email"}}, DoUpdates: clause.AssignmentColumns([]string{"up", "down", "updated_at"}), + Where: clause.Where{Exprs: []clause.Expression{clause.Expr{ + SQL: `(client_global_traffics.up IS NULL + OR client_global_traffics.up <> excluded.up + OR client_global_traffics.down IS NULL + OR client_global_traffics.down <> excluded.down + OR client_global_traffics.updated_at IS NULL + OR client_global_traffics.updated_at <= ?)`, + Vars: []any{heartbeatCutoff}, + }}}, }).Create(&batch).Error; err != nil { return err } @@ -99,17 +106,13 @@ func chunkGlobalRows(rows []model.ClientGlobalTraffic, size int) [][]model.Clien return out } -// overlayGlobalTraffic raises Up/Down on the given rows to the largest global -// value any master still pushing for that email reported. Read-path only — -// callers hand it rows about to be serialized for display; the stored counters -// are untouched. Rows older than globalTrafficFreshWindow are ignored: they -// come from a master that stopped pushing, and folding their frozen counters -// in would keep showing usage the client no longer has (#6113). +// overlayGlobalTraffic applies each direction's largest fresh per-master value for display. +// Stored local counters remain untouched; stale masters cannot inflate usage (#6113). func overlayGlobalTraffic(db *gorm.DB, rows []*xray.ClientTraffic) { if len(rows) == 0 { return } - freshSince := globalTrafficFreshSince() + freshSince := globalTrafficFreshSince(db.NowFunc()) // Cheap short-circuit for the common case (a panel no master pushes to). var probe int64 if err := db.Model(&model.ClientGlobalTraffic{}).