mirror of
https://github.com/MHSanaei/3x-ui.git
synced 2026-08-21 02:21:01 +00:00
f7db247b07
Client CRUD latency scaled with the number of client-inbound edges rather than with the size of the change. On a 5k-client / 8-inbound / ~56k-edge PostgreSQL panel, creating one client took 60-120s (#6252). Two independent causes, both confirmed by the reporter's pg_stat_statements and reproduced locally at their topology. SyncInbound deleted every client_inbounds row for an inbound and re-inserted the whole set, so a one-client edit rewrote thousands of unrelated rows. The dominant caller was not user CRUD: the node traffic poll re-syncs every node inbound from its snapshot every 5s, so the panel churned the entire membership table continuously in the background. SyncInbound now reads the current links and writes only the difference - insert missing, update a changed flow_override, delete departed. Callers are unchanged, so every reconciliation path benefits, and the four hot client CRUD paths additionally pass only the clients they touched via ApplyInboundClientDelta. The insert needs clause.OnConflict: the unconditional delete it replaces also serialized concurrent syncs of one inbound, and the node poll commits in its own transaction outside the serialized writer, where a duplicate key would abort the whole poll on PostgreSQL. Identity and membership questions expanded every inbound's settings.clients JSON - 5.75s per call under the reporter's load. They now read the indexed clients and client_inbounds tables, which every read path already trusts, over just the emails being checked. A LOWER(email) expression index keeps the case-insensitive matching indexed; a struct tag cannot declare one. Measured on PostgreSQL 17 at 8 inbounds x 6000 clients, rows written to client_inbounds per operation, before -> after: create across 8 inbounds 48008 ins / 48000 del -> 8 ins / 0 del update the client 48008 ins / 48008 del -> 0 ins / 0 del detach from 4 inbounds 24000 ins / 24004 del -> 0 ins / 4 del delete the client 24000 ins / 24004 del -> 0 ins / 4 del Two behavior changes worth naming. An email seen with two different subIds across two inbounds' JSON used to be locked so that no add could claim it, including the one with the correct subId; the clients row now adjudicates. And on an install whose settings JSON holds an email with no matching link, "is this email on another inbound" now answers no, so deleting it elsewhere purges its traffic rows; compactOrphans and the startup heal already converge such drift. Every added test was verified against a hand-written mutation of this change, so none of them pass regardless of the fix. One mutation survives on purpose: swapping OnConflict DoUpdates for DoNothing is only observable when two transactions race the same row, and a timing-dependent test would be flaky. Per-node batching of remote pushes and the metadata-only inbounds list from the same report are deliberately not in this change. Closes #6252
532 lines
23 KiB
Go
532 lines
23 KiB
Go
package service
|
|
|
|
import (
|
|
"fmt"
|
|
"path/filepath"
|
|
"strings"
|
|
"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/web/runtime"
|
|
"github.com/mhsanaei/3x-ui/v3/internal/xray"
|
|
)
|
|
|
|
func initTrafficTestDB(t *testing.T) *gorm.DB {
|
|
t.Helper()
|
|
dbDir := t.TempDir()
|
|
t.Setenv("XUI_DB_FOLDER", dbDir)
|
|
if err := database.InitDB(filepath.Join(dbDir, "x-ui.db")); err != nil {
|
|
t.Fatalf("InitDB: %v", err)
|
|
}
|
|
t.Cleanup(func() { _ = database.CloseDB() })
|
|
return database.GetDB()
|
|
}
|
|
|
|
func createNodeInbound(t *testing.T, db *gorm.DB, nodeID int, tag string, port int) {
|
|
t.Helper()
|
|
nid := nodeID
|
|
ib := &model.Inbound{UserId: 1, Tag: tag, Enable: true, Port: port, Protocol: model.VLESS, NodeID: &nid}
|
|
if err := db.Create(ib).Error; err != nil {
|
|
t.Fatalf("create node inbound %q: %v", tag, err)
|
|
}
|
|
}
|
|
|
|
// createNodeInboundWithClient mirrors createNodeInbound but stores the client
|
|
// in the settings JSON, which the node sync turns into a client_inbounds link.
|
|
func createNodeInboundWithClient(t *testing.T, db *gorm.DB, nodeID int, tag string, port int, email string) {
|
|
t.Helper()
|
|
nid := nodeID
|
|
settings := fmt.Sprintf(`{"clients": [{"email": %q, "enable": true}]}`, email)
|
|
ib := &model.Inbound{UserId: 1, Tag: tag, Enable: true, Port: port, Protocol: model.VLESS, NodeID: &nid, Settings: settings}
|
|
if err := db.Create(ib).Error; err != nil {
|
|
t.Fatalf("create node inbound %q: %v", tag, err)
|
|
}
|
|
}
|
|
|
|
func syncNode(t *testing.T, svc *InboundService, nodeID int, tag string, stats ...xray.ClientTraffic) {
|
|
t.Helper()
|
|
snap := &runtime.TrafficSnapshot{
|
|
Inbounds: []*model.Inbound{{Tag: tag, ClientStats: stats}},
|
|
}
|
|
if _, err := svc.setRemoteTrafficLocked(nodeID, snap, false); err != nil {
|
|
t.Fatalf("setRemoteTrafficLocked node %d: %v", nodeID, err)
|
|
}
|
|
}
|
|
|
|
// syncNodeWithSettings mirrors syncNode but carries a real settings JSON on
|
|
// the snapshot inbound, like production nodes do — the sync mirrors snapshot
|
|
// settings onto the central row, and the shared-accumulator guard reads the
|
|
// clients out of those settings.
|
|
func syncNodeWithSettings(t *testing.T, svc *InboundService, nodeID int, tag, settings string, stats ...xray.ClientTraffic) {
|
|
t.Helper()
|
|
snap := &runtime.TrafficSnapshot{
|
|
Inbounds: []*model.Inbound{{Tag: tag, Settings: settings, ClientStats: stats}},
|
|
}
|
|
if _, err := svc.setRemoteTrafficLocked(nodeID, snap, false); err != nil {
|
|
t.Fatalf("setRemoteTrafficLocked node %d: %v", nodeID, err)
|
|
}
|
|
}
|
|
|
|
func readTraffic(t *testing.T, db *gorm.DB, email string) xray.ClientTraffic {
|
|
t.Helper()
|
|
var ct xray.ClientTraffic
|
|
if err := db.Model(xray.ClientTraffic{}).Where("email = ?", email).First(&ct).Error; err != nil {
|
|
t.Fatalf("read client_traffics %q: %v", email, err)
|
|
}
|
|
return ct
|
|
}
|
|
|
|
func assertUpDown(t *testing.T, ct xray.ClientTraffic, wantUp, wantDown int64, when string) {
|
|
t.Helper()
|
|
if ct.Up != wantUp || ct.Down != wantDown {
|
|
t.Errorf("%s: up=%d down=%d, want %d/%d", when, ct.Up, ct.Down, wantUp, wantDown)
|
|
}
|
|
}
|
|
|
|
func TestTwoNodesShareEmail_SumsCorrectly(t *testing.T) {
|
|
db := initTrafficTestDB(t)
|
|
createNodeInbound(t, db, 1, "n1-in", 41001)
|
|
createNodeInbound(t, db, 2, "n2-in", 41002)
|
|
svc := &InboundService{}
|
|
|
|
const email = "shared"
|
|
|
|
// First-ever sync from each node: the row is created at 0 and the current
|
|
// node counters become the baseline. Only deltas beyond those baselines count.
|
|
syncNode(t, svc, 1, "n1-in", xray.ClientTraffic{Email: email, Up: 100, Down: 100, Enable: true})
|
|
syncNode(t, svc, 2, "n2-in", xray.ClientTraffic{Email: email, Up: 200, Down: 200, Enable: true})
|
|
|
|
assertUpDown(t, readTraffic(t, db, email), 0, 0, "after baselines — historical node traffic not imported")
|
|
|
|
syncNode(t, svc, 1, "n1-in", xray.ClientTraffic{Email: email, Up: 150, Down: 150, Enable: true})
|
|
syncNode(t, svc, 2, "n2-in", xray.ClientTraffic{Email: email, Up: 260, Down: 260, Enable: true})
|
|
|
|
assertUpDown(t, readTraffic(t, db, email), 110, 110, "after both nodes grow — deltas (50+60) accrue")
|
|
}
|
|
|
|
func TestSingleNode_MirrorsCorrectly(t *testing.T) {
|
|
db := initTrafficTestDB(t)
|
|
createNodeInbound(t, db, 1, "n1-in", 41001)
|
|
svc := &InboundService{}
|
|
|
|
const email = "solo"
|
|
// First sync: row seeded at 0, current counter becomes the baseline.
|
|
syncNode(t, svc, 1, "n1-in", xray.ClientTraffic{Email: email, Up: 500, Down: 600, Enable: true})
|
|
assertUpDown(t, readTraffic(t, db, email), 0, 0, "first sync — historical traffic not imported")
|
|
|
|
// Second sync: delta (700-500=200 / 800-600=200) accrues normally.
|
|
syncNode(t, svc, 1, "n1-in", xray.ClientTraffic{Email: email, Up: 700, Down: 800, Enable: true})
|
|
assertUpDown(t, readTraffic(t, db, email), 200, 200, "second sync — delta accrues")
|
|
}
|
|
|
|
func TestNodeAdd_ImportsClientHistoryWithNewInbound(t *testing.T) {
|
|
db := initTrafficTestDB(t)
|
|
svc := &InboundService{}
|
|
|
|
const email = "newnode-client"
|
|
const histUp, histDown int64 = 6_000_000_000, 200_000_000_000
|
|
|
|
syncNode(t, svc, 1, "fresh-in", xray.ClientTraffic{Email: email, Up: histUp, Down: histDown, Enable: true})
|
|
assertUpDown(t, readTraffic(t, db, email), histUp, histDown, "node-add: client history imported with its brand-new inbound")
|
|
|
|
syncNode(t, svc, 1, "fresh-in", xray.ClientTraffic{Email: email, Up: histUp + 1024, Down: histDown + 2048, Enable: true})
|
|
assertUpDown(t, readTraffic(t, db, email), histUp+1024, histDown+2048, "post-import delta accrues, no double count")
|
|
}
|
|
|
|
// TestStaleSnapshot_DeletedClientNotResurrected reproduces #5739: a snapshot
|
|
// fetched just before a client's deletion still names the email. Applying it
|
|
// must neither recreate the client_traffics row (at zero) nor re-add the
|
|
// client to the central inbound's settings while the delete tombstone lives.
|
|
func TestStaleSnapshot_DeletedClientNotResurrected(t *testing.T) {
|
|
db := initTrafficTestDB(t)
|
|
createNodeInboundWithClient(t, db, 1, "n1-in", 41001, "victim-5739")
|
|
svc := &InboundService{}
|
|
|
|
const email = "victim-5739"
|
|
withClient := fmt.Sprintf(`{"clients": [{"email": %q, "enable": true}]}`, email)
|
|
syncNodeWithSettings(t, svc, 1, "n1-in", withClient, xray.ClientTraffic{Email: email, Up: 100, Down: 100, Enable: true})
|
|
|
|
if err := db.Model(&model.Inbound{}).Where("tag = ?", "n1-in").
|
|
Update("settings", `{"clients": []}`).Error; err != nil {
|
|
t.Fatalf("clear central settings: %v", err)
|
|
}
|
|
if err := db.Where("email = ?", email).Delete(&xray.ClientTraffic{}).Error; err != nil {
|
|
t.Fatalf("delete stats row: %v", err)
|
|
}
|
|
tombstoneClientEmail(email)
|
|
|
|
syncNodeWithSettings(t, svc, 1, "n1-in", withClient, xray.ClientTraffic{Email: email, Up: 120, Down: 120, Enable: true})
|
|
|
|
var rows int64
|
|
if err := db.Model(xray.ClientTraffic{}).Where("email = ?", email).Count(&rows).Error; err != nil {
|
|
t.Fatalf("count stats rows: %v", err)
|
|
}
|
|
if rows != 0 {
|
|
t.Errorf("deleted client's stats row resurrected by stale snapshot (%d rows)", rows)
|
|
}
|
|
|
|
var ib model.Inbound
|
|
if err := db.Where("tag = ?", "n1-in").First(&ib).Error; err != nil {
|
|
t.Fatalf("load inbound: %v", err)
|
|
}
|
|
if strings.Contains(ib.Settings, email) {
|
|
t.Errorf("deleted client re-added to central settings: %s", ib.Settings)
|
|
}
|
|
}
|
|
|
|
func TestNodeAdd_TombstonedClientNotResurrected(t *testing.T) {
|
|
db := initTrafficTestDB(t)
|
|
svc := &InboundService{}
|
|
|
|
const email = "deleted-ghost"
|
|
const stale int64 = 50_000_000_000
|
|
|
|
tombstoneClientEmail(email)
|
|
syncNode(t, svc, 1, "fresh-in", xray.ClientTraffic{Email: email, Up: stale, Down: stale, Enable: true})
|
|
assertUpDown(t, readTraffic(t, db, email), 0, 0, "tombstoned client must not resurrect via node-add seed")
|
|
}
|
|
|
|
func TestUpgrade_PreExistingRow_NoDoubleCount(t *testing.T) {
|
|
db := initTrafficTestDB(t)
|
|
createNodeInbound(t, db, 1, "n1-in", 41001)
|
|
svc := &InboundService{}
|
|
|
|
const email = "legacy"
|
|
var ib model.Inbound
|
|
if err := db.Where("tag = ?", "n1-in").First(&ib).Error; err != nil {
|
|
t.Fatalf("load inbound: %v", err)
|
|
}
|
|
if err := db.Create(&xray.ClientTraffic{InboundId: ib.Id, Email: email, Up: 1000, Down: 2000, Enable: true}).Error; err != nil {
|
|
t.Fatalf("seed pre-existing row: %v", err)
|
|
}
|
|
|
|
syncNode(t, svc, 1, "n1-in", xray.ClientTraffic{Email: email, Up: 1000, Down: 2000, Enable: true})
|
|
assertUpDown(t, readTraffic(t, db, email), 1000, 2000, "first snapshot must not double-count")
|
|
|
|
syncNode(t, svc, 1, "n1-in", xray.ClientTraffic{Email: email, Up: 1100, Down: 2100, Enable: true})
|
|
assertUpDown(t, readTraffic(t, db, email), 1100, 2100, "growth after upgrade accrues")
|
|
}
|
|
|
|
// TestGhostData_NoPhantomTraffic reproduces the 50 GB phantom traffic bug:
|
|
// a node retains stale data for a deleted client (e.g. de-1 was offline during
|
|
// deletion). When the master first syncs this email again it must NOT import the
|
|
// stale counter — it sets Up=0 and records the current node value as the
|
|
// baseline so only future deltas count.
|
|
func TestGhostData_NoPhantomTraffic(t *testing.T) {
|
|
db := initTrafficTestDB(t)
|
|
createNodeInbound(t, db, 1, "de1-in", 41001)
|
|
svc := &InboundService{}
|
|
|
|
const email = "pouya"
|
|
const staleBytes int64 = 50 * 1024 * 1024 * 1024 // 50 GB stale on de-1
|
|
|
|
// Node reports a client with a large pre-existing counter (ghost data from a
|
|
// deleted account that survived on the node). Master has no row for this email.
|
|
syncNode(t, svc, 1, "de1-in", xray.ClientTraffic{Email: email, Up: staleBytes, Down: staleBytes, Enable: true})
|
|
|
|
ct := readTraffic(t, db, email)
|
|
if ct.Up >= staleBytes || ct.Down >= staleBytes {
|
|
t.Errorf("ghost data imported: up=%d down=%d; stale node counter must not count as new traffic", ct.Up, ct.Down)
|
|
}
|
|
assertUpDown(t, ct, 0, 0, "first sync of ghost email — imported as 0, not 50 GB")
|
|
|
|
// Subsequent syncs only add incremental usage beyond the baseline.
|
|
syncNode(t, svc, 1, "de1-in", xray.ClientTraffic{Email: email, Up: staleBytes + 1024, Down: staleBytes + 2048, Enable: true})
|
|
assertUpDown(t, readTraffic(t, db, email), 1024, 2048, "only incremental traffic beyond baseline counts")
|
|
}
|
|
|
|
func TestNodeCounterReset_NoReAdd(t *testing.T) {
|
|
db := initTrafficTestDB(t)
|
|
createNodeInbound(t, db, 1, "n1-in", 41001)
|
|
svc := &InboundService{}
|
|
|
|
const email = "restart"
|
|
// First sync seeds at 0 (baseline=900). Second sync adds delta 950-900=50.
|
|
syncNode(t, svc, 1, "n1-in", xray.ClientTraffic{Email: email, Up: 900, Down: 900, Enable: true})
|
|
syncNode(t, svc, 1, "n1-in", xray.ClientTraffic{Email: email, Up: 950, Down: 950, Enable: true})
|
|
assertUpDown(t, readTraffic(t, db, email), 50, 50, "before node reset")
|
|
|
|
// Node reboot drops the counter to 50. delta=50-950=-900 is a counter reset,
|
|
// not new traffic: add 0 and rebaseline to 50, never re-add the node's full
|
|
// cumulative counter onto the master total (#5456).
|
|
syncNode(t, svc, 1, "n1-in", xray.ClientTraffic{Email: email, Up: 50, Down: 50, Enable: true})
|
|
ct := readTraffic(t, db, email)
|
|
if ct.Up < 0 || ct.Down < 0 {
|
|
t.Fatalf("row went negative after node reset: up=%d down=%d", ct.Up, ct.Down)
|
|
}
|
|
assertUpDown(t, ct, 50, 50, "after node counter reset: rebaselined, not re-added")
|
|
|
|
// Post-reset accrual resumes from the new baseline: 80-50=30.
|
|
syncNode(t, svc, 1, "n1-in", xray.ClientTraffic{Email: email, Up: 80, Down: 80, Enable: true})
|
|
assertUpDown(t, readTraffic(t, db, email), 80, 80, "post-reset delta accrues from rebaselined counter")
|
|
}
|
|
|
|
func TestCentralReset_NoReAdd(t *testing.T) {
|
|
db := initTrafficTestDB(t)
|
|
createNodeInbound(t, db, 1, "n1-in", 41001)
|
|
createNodeInbound(t, db, 2, "n2-in", 41002)
|
|
svc := &InboundService{}
|
|
|
|
const email = "reset"
|
|
syncNode(t, svc, 1, "n1-in", xray.ClientTraffic{Email: email, Up: 100, Down: 100, Enable: true})
|
|
syncNode(t, svc, 2, "n2-in", xray.ClientTraffic{Email: email, Up: 100, Down: 100, Enable: true})
|
|
syncNode(t, svc, 1, "n1-in", xray.ClientTraffic{Email: email, Up: 200, Down: 200, Enable: true})
|
|
syncNode(t, svc, 2, "n2-in", xray.ClientTraffic{Email: email, Up: 200, Down: 200, Enable: true})
|
|
|
|
if err := db.Model(xray.ClientTraffic{}).Where("email = ?", email).
|
|
Updates(map[string]any{"up": 0, "down": 0}).Error; err != nil {
|
|
t.Fatalf("simulate central reset: %v", err)
|
|
}
|
|
|
|
syncNode(t, svc, 1, "n1-in", xray.ClientTraffic{Email: email, Up: 210, Down: 210, Enable: true})
|
|
syncNode(t, svc, 2, "n2-in", xray.ClientTraffic{Email: email, Up: 205, Down: 205, Enable: true})
|
|
|
|
assertUpDown(t, readTraffic(t, db, email), 15, 15, "after central reset only increments accrue")
|
|
}
|
|
|
|
// A real reset (ResetClientTrafficByEmail) must clear the per-node baseline so
|
|
// the node's pre-reset cumulative — including traffic it counted but had not yet
|
|
// synced — cannot leak back onto the master after the reset (#5476, #5390).
|
|
func TestCentralResetClearsNodeBaseline_NoLeak(t *testing.T) {
|
|
db := initTrafficTestDB(t)
|
|
createNodeInbound(t, db, 1, "n1-in", 41001)
|
|
StartTrafficWriter()
|
|
svc := &InboundService{}
|
|
|
|
const email = "reset-revert"
|
|
syncNode(t, svc, 1, "n1-in", xray.ClientTraffic{Email: email, Up: 100, Down: 100, Enable: true})
|
|
syncNode(t, svc, 1, "n1-in", xray.ClientTraffic{Email: email, Up: 300, Down: 300, Enable: true})
|
|
assertUpDown(t, readTraffic(t, db, email), 200, 200, "before reset")
|
|
|
|
if err := svc.ResetClientTrafficByEmail(email); err != nil {
|
|
t.Fatalf("ResetClientTrafficByEmail: %v", err)
|
|
}
|
|
assertUpDown(t, readTraffic(t, db, email), 0, 0, "right after reset")
|
|
|
|
var baselines int64
|
|
if err := db.Model(&model.NodeClientTraffic{}).Where("email = ?", email).Count(&baselines).Error; err != nil {
|
|
t.Fatalf("count baselines: %v", err)
|
|
}
|
|
if baselines != 0 {
|
|
t.Fatalf("reset must clear node baseline rows, found %d", baselines)
|
|
}
|
|
|
|
// Node still reports its pre-reset cumulative (340 > last synced 300: usage it
|
|
// had not synced before the reset). It must not revert the reset.
|
|
syncNode(t, svc, 1, "n1-in", xray.ClientTraffic{Email: email, Up: 340, Down: 340, Enable: true})
|
|
assertUpDown(t, readTraffic(t, db, email), 0, 0, "stale node counter must not revert reset")
|
|
|
|
// Genuine post-reset usage accrues from the rebaselined counter: 370-340=30.
|
|
syncNode(t, svc, 1, "n1-in", xray.ClientTraffic{Email: email, Up: 370, Down: 370, Enable: true})
|
|
assertUpDown(t, readTraffic(t, db, email), 30, 30, "post-reset usage accrues")
|
|
}
|
|
|
|
func TestInboundRemoval_KeepsSharedEmailRow(t *testing.T) {
|
|
db := initTrafficTestDB(t)
|
|
createNodeInboundWithClient(t, db, 1, "n1-in", 41001, "shared")
|
|
createNodeInboundWithClient(t, db, 2, "n2-in", 41002, "shared")
|
|
svc := &InboundService{}
|
|
|
|
const email = "shared"
|
|
settings := fmt.Sprintf(`{"clients": [{"email": %q, "enable": true}]}`, email)
|
|
// Baselines set: node1=100, node2=200. Row seeded at 0.
|
|
syncNodeWithSettings(t, svc, 1, "n1-in", settings, xray.ClientTraffic{Email: email, Up: 100, Down: 100, Enable: true})
|
|
syncNodeWithSettings(t, svc, 2, "n2-in", settings, xray.ClientTraffic{Email: email, Up: 200, Down: 200, Enable: true})
|
|
// node1 delta=50, node2 delta=60 → total=110.
|
|
syncNodeWithSettings(t, svc, 1, "n1-in", settings, xray.ClientTraffic{Email: email, Up: 150, Down: 150, Enable: true})
|
|
syncNodeWithSettings(t, svc, 2, "n2-in", settings, xray.ClientTraffic{Email: email, Up: 260, Down: 260, Enable: true})
|
|
assertUpDown(t, readTraffic(t, db, email), 110, 110, "baseline sum")
|
|
|
|
// Node 1 rebuilt (reinstall / another master's reconcile): its inbound
|
|
// vanishes from the snapshot. The shared accumulator must survive — losing
|
|
// it would let the next node sync re-seed the row with that node's counter
|
|
// alone, showing only the last panel's number instead of the sum.
|
|
if _, err := svc.setRemoteTrafficLocked(1, &runtime.TrafficSnapshot{}, false); err != nil {
|
|
t.Fatalf("sync node 1 with empty snapshot: %v", err)
|
|
}
|
|
assertUpDown(t, readTraffic(t, db, email), 110, 110, "after node 1 inbound removal")
|
|
|
|
// Node 2 keeps accruing onto the surviving row. node2 delta=300-260=40 → 150.
|
|
syncNodeWithSettings(t, svc, 2, "n2-in", settings, xray.ClientTraffic{Email: email, Up: 300, Down: 300, Enable: true})
|
|
assertUpDown(t, readTraffic(t, db, email), 150, 150, "after node 2 grows")
|
|
}
|
|
|
|
func TestClientGoneFromOneNode_KeepsSharedEmailRow(t *testing.T) {
|
|
db := initTrafficTestDB(t)
|
|
createNodeInboundWithClient(t, db, 1, "n1-in", 41001, "shared")
|
|
createNodeInboundWithClient(t, db, 2, "n2-in", 41002, "shared")
|
|
svc := &InboundService{}
|
|
|
|
const email = "shared"
|
|
settings := fmt.Sprintf(`{"clients": [{"email": %q, "enable": true}]}`, email)
|
|
// First syncs set baselines (node1=100, node2=200). Row seeded at 0.
|
|
syncNodeWithSettings(t, svc, 1, "n1-in", settings, xray.ClientTraffic{Email: email, Up: 100, Down: 100, Enable: true})
|
|
syncNodeWithSettings(t, svc, 2, "n2-in", settings, xray.ClientTraffic{Email: email, Up: 200, Down: 200, Enable: true})
|
|
|
|
// Client detached from node 1's inbound only: its stats vanish from that
|
|
// inbound's snapshot while node 2 still hosts the email.
|
|
syncNodeWithSettings(t, svc, 1, "n1-in", `{"clients": []}`)
|
|
assertUpDown(t, readTraffic(t, db, email), 0, 0, "after client left node 1 — row unchanged at 0")
|
|
|
|
// Node 2 delta: 240-200=40 → accrues to 40.
|
|
syncNodeWithSettings(t, svc, 2, "n2-in", settings, xray.ClientTraffic{Email: email, Up: 240, Down: 240, Enable: true})
|
|
assertUpDown(t, readTraffic(t, db, email), 40, 40, "node 2 keeps accruing")
|
|
}
|
|
|
|
// TestStatsUnderSiblingInbound_KeepsNodeBaseline reproduces the recurring
|
|
// sweep bug behind #5202: the client is attached to two inbounds of the SAME
|
|
// node, the node reports its stats under n1-a only, but the master-side row
|
|
// is owned by n1-b's mirror. The per-email sweep for n1-b must not drop the
|
|
// (node, email) baseline that n1-a's delta computation needs — doing so every
|
|
// cycle froze the client's counter permanently.
|
|
func TestStatsUnderSiblingInbound_KeepsNodeBaseline(t *testing.T) {
|
|
db := initTrafficTestDB(t)
|
|
createNodeInboundWithClient(t, db, 1, "n1-a", 41001, "fresh")
|
|
createNodeInbound(t, db, 1, "n1-b", 41002)
|
|
svc := &InboundService{}
|
|
|
|
const email = "fresh"
|
|
var ibB model.Inbound
|
|
if err := db.Where("tag = ?", "n1-b").First(&ibB).Error; err != nil {
|
|
t.Fatalf("load n1-b: %v", err)
|
|
}
|
|
// Master-side row created when the client was added on the panel, owned by
|
|
// n1-b's mirror (e.g. the client form targeted that inbound).
|
|
if err := db.Create(&xray.ClientTraffic{InboundId: ibB.Id, Email: email, Enable: true}).Error; err != nil {
|
|
t.Fatalf("seed master row: %v", err)
|
|
}
|
|
|
|
settings := fmt.Sprintf(`{"clients": [{"email": %q, "enable": true}]}`, email)
|
|
sync := func(up, down int64) {
|
|
t.Helper()
|
|
snap := &runtime.TrafficSnapshot{Inbounds: []*model.Inbound{
|
|
{Tag: "n1-a", Settings: settings, ClientStats: []xray.ClientTraffic{{Email: email, Up: up, Down: down, Enable: true}}},
|
|
{Tag: "n1-b", Settings: `{"clients": []}`},
|
|
}}
|
|
if _, err := svc.setRemoteTrafficLocked(1, snap, false); err != nil {
|
|
t.Fatalf("sync: %v", err)
|
|
}
|
|
}
|
|
|
|
sync(630, 630)
|
|
var baselines int64
|
|
if err := db.Model(&model.NodeClientTraffic{}).Where("node_id = ? AND email = ?", 1, email).Count(&baselines).Error; err != nil {
|
|
t.Fatalf("count baselines: %v", err)
|
|
}
|
|
if baselines != 1 {
|
|
t.Fatalf("baseline must survive the sibling-inbound sweep, found %d rows", baselines)
|
|
}
|
|
|
|
sync(700, 700)
|
|
assertUpDown(t, readTraffic(t, db, email), 70, 70, "delta accrues once baseline survives")
|
|
}
|
|
|
|
// TestMultiAttach_SameNode_DivergentSiblings reproduces #5274: a client is
|
|
// attached to several inbounds of the SAME node. Xray reports client traffic
|
|
// globally per email, so the node's enriched inbound list copies one shared
|
|
// counter onto every inbound the client is on. When those copies diverge — a
|
|
// legacy per-inbound row surviving the v3.2.x→v3.3.x upgrade, or any drift —
|
|
// the per-inbound delta loop used to treat the lower sibling as a node-counter
|
|
// reset and re-add its full value, inflating the client far past real usage.
|
|
// The merge must collapse the email to its node-wide total and count it once.
|
|
func TestMultiAttach_SameNode_DivergentSiblings(t *testing.T) {
|
|
db := initTrafficTestDB(t)
|
|
createNodeInboundWithClient(t, db, 1, "n1-a", 41001, "multi")
|
|
createNodeInboundWithClient(t, db, 1, "n1-b", 41002, "multi")
|
|
createNodeInboundWithClient(t, db, 1, "n1-c", 41003, "multi")
|
|
svc := &InboundService{}
|
|
|
|
const email = "multi"
|
|
settings := fmt.Sprintf(`{"clients": [{"email": %q, "enable": true}]}`, email)
|
|
|
|
// The three inbounds report the same email with diverging values; the
|
|
// node's true per-email total is the largest (the shared global counter).
|
|
sync := func(a, b, c int64) {
|
|
t.Helper()
|
|
snap := &runtime.TrafficSnapshot{Inbounds: []*model.Inbound{
|
|
{Tag: "n1-a", Settings: settings, ClientStats: []xray.ClientTraffic{{Email: email, Up: a, Down: a, Enable: true}}},
|
|
{Tag: "n1-b", Settings: settings, ClientStats: []xray.ClientTraffic{{Email: email, Up: b, Down: b, Enable: true}}},
|
|
{Tag: "n1-c", Settings: settings, ClientStats: []xray.ClientTraffic{{Email: email, Up: c, Down: c, Enable: true}}},
|
|
}}
|
|
if _, err := svc.setRemoteTrafficLocked(1, snap, false); err != nil {
|
|
t.Fatalf("sync: %v", err)
|
|
}
|
|
}
|
|
|
|
// First-ever sync: row seeded at 0, canon (max of siblings = 100) becomes the
|
|
// baseline. The node total is NOT imported as historical traffic — this prevents
|
|
// ghost data from a previously-deleted account being counted as real usage.
|
|
sync(100, 50, 80)
|
|
assertUpDown(t, readTraffic(t, db, email), 0, 0, "first sync seeds at 0 — not the sum (230) nor the max (100)")
|
|
|
|
// Second sync: canon grew from 100→150, delta=50 accrues. Siblings 60 and 90
|
|
// do not re-add their full values — the canon clamp prevents inflation.
|
|
sync(150, 60, 90)
|
|
assertUpDown(t, readTraffic(t, db, email), 50, 50, "second sync: only the 50-unit increment counts, not every sibling")
|
|
|
|
// Equal siblings (the healthy current-schema case): canon grew from 150→200,
|
|
// delta=50 accrues once.
|
|
sync(200, 200, 200)
|
|
assertUpDown(t, readTraffic(t, db, email), 100, 100, "equal siblings accrue the single increment")
|
|
}
|
|
|
|
func TestDelClientStat_CleansNodeBaselines(t *testing.T) {
|
|
db := initTrafficTestDB(t)
|
|
svc := &InboundService{}
|
|
|
|
const email = "gone"
|
|
if err := db.Create(&xray.ClientTraffic{InboundId: 1, Email: email, Enable: true}).Error; err != nil {
|
|
t.Fatalf("seed client_traffics: %v", err)
|
|
}
|
|
if err := db.Create(&model.NodeClientTraffic{NodeId: 1, Email: email, Up: 10, Down: 10}).Error; err != nil {
|
|
t.Fatalf("seed node baseline 1: %v", err)
|
|
}
|
|
if err := db.Create(&model.NodeClientTraffic{NodeId: 2, Email: email, Up: 20, Down: 20}).Error; err != nil {
|
|
t.Fatalf("seed node baseline 2: %v", err)
|
|
}
|
|
|
|
if err := svc.DelClientStat(db, email); err != nil {
|
|
t.Fatalf("DelClientStat: %v", err)
|
|
}
|
|
|
|
var cnt int64
|
|
if err := db.Model(&model.NodeClientTraffic{}).Where("email = ?", email).Count(&cnt).Error; err != nil {
|
|
t.Fatalf("count baselines: %v", err)
|
|
}
|
|
if cnt != 0 {
|
|
t.Errorf("expected node baselines cleaned, found %d", cnt)
|
|
}
|
|
}
|
|
|
|
func TestNodeDelete_CleansNodeBaselines(t *testing.T) {
|
|
db := initTrafficTestDB(t)
|
|
nodeSvc := NodeService{}
|
|
|
|
if err := db.Create(&model.NodeClientTraffic{NodeId: 7, Email: "a", Up: 1, Down: 1}).Error; err != nil {
|
|
t.Fatalf("seed node 7 a: %v", err)
|
|
}
|
|
if err := db.Create(&model.NodeClientTraffic{NodeId: 7, Email: "b", Up: 2, Down: 2}).Error; err != nil {
|
|
t.Fatalf("seed node 7 b: %v", err)
|
|
}
|
|
if err := db.Create(&model.NodeClientTraffic{NodeId: 8, Email: "c", Up: 3, Down: 3}).Error; err != nil {
|
|
t.Fatalf("seed node 8 c: %v", err)
|
|
}
|
|
|
|
if err := nodeSvc.Delete(7); err != nil {
|
|
t.Fatalf("NodeService.Delete(7): %v", err)
|
|
}
|
|
|
|
var sevenCnt, eightCnt int64
|
|
db.Model(&model.NodeClientTraffic{}).Where("node_id = ?", 7).Count(&sevenCnt)
|
|
db.Model(&model.NodeClientTraffic{}).Where("node_id = ?", 8).Count(&eightCnt)
|
|
if sevenCnt != 0 {
|
|
t.Errorf("node 7 baselines not cleaned: %d remain", sevenCnt)
|
|
}
|
|
if eightCnt != 1 {
|
|
t.Errorf("node 8 baseline should survive, found %d", eightCnt)
|
|
}
|
|
}
|