From fb03b0e9f1c0ef8329412439c5b30771033a4467 Mon Sep 17 00:00:00 2001 From: Younes <62533356+younesvatan78@users.noreply.github.com> Date: Sat, 20 Jun 2026 02:06:35 +0330 Subject: [PATCH] fix(traffic): prevent phantom quota consumption from stale node data (#5412) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three related bugs caused inflated traffic counters and spurious quota hits on multi-node setups, most visibly when a client email was renamed while a node was offline or its PostgreSQL deadlocked. **Fix 1 — phantom quota (root cause)** `setRemoteTrafficLocked` new-row path: when master had no `client_traffics` row for an email that a node reported, it seeded the row with `Up: cs.Up` — importing the node's full accumulated counter as if it were fresh quota usage. If the node retained stale data from a previously-deleted account (e.g. a failed deletion during an outage), the ghost 50 GB appeared on the new client immediately and triggered `disableInvalidClients` the same tick. Fixed by seeding at `Up: 0`; the current node value still becomes the baseline so only future increments count. **Fix 2 — PostgreSQL deadlock** `addClientTraffic` did a read-modify-write via `tx.Save(slice)`, issuing UPDATEs in slice order. Two concurrent goroutines locking the same rows in opposite order deadlock on PostgreSQL (SQLite avoids this with file-level serialisation). Replaced with atomic per-email `UPDATE SET up=up+?, down=down+?` statements. Also preserves the delayed-start ExpiryTime conversion that `adjustTraffics` computes in-memory but the old Save path persisted to the DB. **Fix 3 & 4 — stale `inbound_id` filters** `autoRenewClients` used `WHERE inbound_id NOT IN (node inbounds)` to skip node clients, but `client_traffics.inbound_id` is set once on INSERT and never refreshed. Replaced with an email-based subquery through `client_inbounds` (the authoritative source). Also added a safe type assertion for `settings["clients"].([]any)` that previously panicked on nil. **Fix 5 — stale `inbound_id` in reset** `resetAllClientTrafficsLocked` used `WHERE inbound_id = ?` to find which emails to reset; same staleness problem. Replaced with the `client_inbounds` join for email lookup; the `inbounds.last_traffic_reset_time` update still correctly uses the inbound ID directly on the `inbounds` table. Tests updated to reflect the new seeding-at-zero semantics and a new `TestGhostData_NoPhantomTraffic` test reproduces the exact 50 GB phantom scenario. --- internal/web/service/client_traffic.go | 31 +++++--- internal/web/service/inbound_node.go | 4 +- internal/web/service/inbound_traffic.go | 58 ++++++++++---- .../service/node_client_traffic_sum_test.go | 76 +++++++++++++++---- 4 files changed, 124 insertions(+), 45 deletions(-) diff --git a/internal/web/service/client_traffic.go b/internal/web/service/client_traffic.go index a5b9e0abb..ee9bc3568 100644 --- a/internal/web/service/client_traffic.go +++ b/internal/web/service/client_traffic.go @@ -121,22 +121,29 @@ func (s *ClientService) resetAllClientTrafficsLocked(id int) error { now := time.Now().Unix() * 1000 if err := db.Transaction(func(tx *gorm.DB) error { - whereText := "inbound_id " - if id == -1 { - whereText += " > ?" - } else { - whereText += " = ?" - } - + // client_traffics.inbound_id is stale: it reflects the inbound the row was + // first inserted under and is never refreshed. Use the client_inbounds join + // as the authoritative source for which emails belong to a given inbound. var resetEmails []string - if err := tx.Model(xray.ClientTraffic{}). - Where(whereText, id). - Pluck("email", &resetEmails).Error; err != nil { - return err + if id == -1 { + if err := tx.Model(xray.ClientTraffic{}).Pluck("email", &resetEmails).Error; err != nil { + return err + } + } else { + if err := tx.Table("client_inbounds ci"). + Select("c.email"). + Joins("JOIN clients c ON c.id = ci.client_id"). + Where("ci.inbound_id = ?", id). + Pluck("c.email", &resetEmails).Error; err != nil { + return err + } + } + if len(resetEmails) == 0 { + return nil } result := tx.Model(xray.ClientTraffic{}). - Where(whereText, id). + Where("email IN ?", resetEmails). Updates(map[string]any{"enable": true, "up": 0, "down": 0}) if result.Error != nil { diff --git a/internal/web/service/inbound_node.go b/internal/web/service/inbound_node.go index 822b24296..aeeab2eff 100644 --- a/internal/web/service/inbound_node.go +++ b/internal/web/service/inbound_node.go @@ -559,8 +559,8 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi Total: cs.Total, ExpiryTime: cs.ExpiryTime, Reset: cs.Reset, - Up: canon.Up, - Down: canon.Down, + Up: 0, + Down: 0, LastOnline: cs.LastOnline, } if err := tx.Clauses(clause.OnConflict{Columns: []clause.Column{{Name: "email"}}, DoNothing: true}). diff --git a/internal/web/service/inbound_traffic.go b/internal/web/service/inbound_traffic.go index 76a4771d9..244f7ad99 100644 --- a/internal/web/service/inbound_traffic.go +++ b/internal/web/service/inbound_traffic.go @@ -133,9 +133,7 @@ func (s *InboundService) addClientTraffic(tx *gorm.DB, traffics []*xray.ClientTr return err } - // Index by email for O(N) merge — the previous nested loop was O(N²) - // and dominated each cron tick on inbounds with thousands of active - // clients (7500 × 7500 = 56M string comparisons every 10 seconds). + // Index by email for O(N) merge. trafficByEmail := make(map[string]*xray.ClientTraffic, len(traffics)) for i := range traffics { if traffics[i] != nil { @@ -143,21 +141,39 @@ func (s *InboundService) addClientTraffic(tx *gorm.DB, traffics []*xray.ClientTr } } now := time.Now().UnixMilli() - for dbTraffic_index := range dbClientTraffics { - t, ok := trafficByEmail[dbClientTraffics[dbTraffic_index].Email] - if !ok { + // Use atomic per-row UPDATE instead of read-modify-write Save. tx.Save + // issues UPDATEs in slice order, which varies between concurrent callers; + // on PostgreSQL two transactions locking the same rows in opposite order + // deadlock. An atomic "SET up = up + ?" never holds a row lock across a + // subsequent lock acquisition, so concurrent writers cannot deadlock. + for _, ct := range dbClientTraffics { + t, ok := trafficByEmail[ct.Email] + if !ok || (t.Up == 0 && t.Down == 0) { continue } - dbClientTraffics[dbTraffic_index].Up += t.Up - dbClientTraffics[dbTraffic_index].Down += t.Down - if t.Up+t.Down > 0 { - dbClientTraffics[dbTraffic_index].LastOnline = now + if err = tx.Exec( + fmt.Sprintf( + `UPDATE client_traffics SET up = up + ?, down = down + ?, last_online = %s WHERE email = ?`, + database.GreatestExpr("last_online", "?"), + ), + t.Up, t.Down, now, ct.Email, + ).Error; err != nil { + logger.Warning("AddClientTraffic update data ", err) } } - err = tx.Save(dbClientTraffics).Error - if err != nil { - logger.Warning("AddClientTraffic update data ", err) + // adjustTraffics converts delayed-start rows (negative ExpiryTime → absolute + // deadline) in-memory. Persist that conversion now since the traffic UPDATE + // above only touches up/down/last_online. + for _, ct := range dbClientTraffics { + if ct.ExpiryTime > 0 { + if err = tx.Exec( + `UPDATE client_traffics SET expiry_time = ? WHERE email = ? AND expiry_time < 0`, + ct.ExpiryTime, ct.Email, + ).Error; err != nil { + logger.Warning("AddClientTraffic update expiry_time ", err) + } + } } return nil @@ -272,9 +288,18 @@ func (s *InboundService) autoRenewClients(tx *gorm.DB) (bool, int64, error) { now := time.Now().Unix() * 1000 var err, err1 error + // Filter to clients that have at least one local inbound. Using + // client_traffics.inbound_id is wrong: it goes stale after an inbound is + // deleted/recreated and always points to the first inbound the client was + // attached to, so it could be a node inbound even when the client also has + // local inbounds. The email-based join through client_inbounds is authoritative. err = tx.Model(xray.ClientTraffic{}). Where("reset > 0 and expiry_time > 0 and expiry_time <= ?", now). - Where("inbound_id NOT IN (?)", tx.Model(&model.Inbound{}).Select("id").Where("node_id IS NOT NULL")). + Where("email IN (?)", tx.Table("client_inbounds ci"). + Select("c.email"). + Joins("JOIN clients c ON c.id = ci.client_id"). + Joins("JOIN inbounds i ON i.id = ci.inbound_id"). + Where("i.node_id IS NULL")). Find(&traffics).Error if err != nil { return false, 0, err @@ -326,7 +351,10 @@ func (s *InboundService) autoRenewClients(tx *gorm.DB) (bool, int64, error) { for inbound_index := range inbounds { settings := map[string]any{} json.Unmarshal([]byte(inbounds[inbound_index].Settings), &settings) - clients := settings["clients"].([]any) + clients, _ := settings["clients"].([]any) + if len(clients) == 0 { + continue + } for client_index := range clients { c := clients[client_index].(map[string]any) for traffic_index, traffic := range traffics { diff --git a/internal/web/service/node_client_traffic_sum_test.go b/internal/web/service/node_client_traffic_sum_test.go index 6d75277f9..0c5dc0b8b 100644 --- a/internal/web/service/node_client_traffic_sum_test.go +++ b/internal/web/service/node_client_traffic_sum_test.go @@ -92,15 +92,17 @@ func TestTwoNodesShareEmail_SumsCorrectly(t *testing.T) { 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), 100, 100, "after baselines") + 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), 210, 210, "after both nodes grow") + assertUpDown(t, readTraffic(t, db, email), 110, 110, "after both nodes grow — deltas (50+60) accrue") } func TestSingleNode_MirrorsCorrectly(t *testing.T) { @@ -109,11 +111,13 @@ func TestSingleNode_MirrorsCorrectly(t *testing.T) { 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), 500, 600, "first sync") + 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), 700, 800, "second sync mirrors cumulative") + assertUpDown(t, readTraffic(t, db, email), 200, 200, "second sync — delta accrues") } func TestUpgrade_PreExistingRow_NoDoubleCount(t *testing.T) { @@ -137,22 +141,52 @@ func TestUpgrade_PreExistingRow_NoDoubleCount(t *testing.T) { 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_Clamped(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), 950, 950, "before node reset") + assertUpDown(t, readTraffic(t, db, email), 50, 50, "before node reset") + // Counter resets to 50 (Xray restart). delta=50-950=-900 → clamped → adds 50. 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, 1000, 1000, "after node counter reset (clamped)") + assertUpDown(t, ct, 100, 100, "after node counter reset (clamped)") } func TestCentralReset_NoReAdd(t *testing.T) { @@ -186,11 +220,13 @@ func TestInboundRemoval_KeepsSharedEmailRow(t *testing.T) { 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), 210, 210, "baseline sum") + 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 @@ -199,11 +235,11 @@ func TestInboundRemoval_KeepsSharedEmailRow(t *testing.T) { 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), 210, 210, "after node 1 inbound removal") + assertUpDown(t, readTraffic(t, db, email), 110, 110, "after node 1 inbound removal") - // Node 2 keeps accruing onto the surviving row. + // 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), 250, 250, "after node 2 grows") + assertUpDown(t, readTraffic(t, db, email), 150, 150, "after node 2 grows") } func TestClientGoneFromOneNode_KeepsSharedEmailRow(t *testing.T) { @@ -214,16 +250,18 @@ func TestClientGoneFromOneNode_KeepsSharedEmailRow(t *testing.T) { 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), 100, 100, "after client left node 1") + 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), 140, 140, "node 2 keeps accruing") + assertUpDown(t, readTraffic(t, db, email), 40, 40, "node 2 keeps accruing") } // TestStatsUnderSiblingInbound_KeepsNodeBaseline reproduces the recurring @@ -306,15 +344,21 @@ func TestMultiAttach_SameNode_DivergentSiblings(t *testing.T) { } } + // 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), 100, 100, "first sync counts the node total once, not the sum") + 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), 150, 150, "second sync: grew by 50, not by every sibling") + 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) must still accrue once. + // 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), 200, 200, "equal siblings accrue the single increment") + assertUpDown(t, readTraffic(t, db, email), 100, 100, "equal siblings accrue the single increment") } func TestDelClientStat_CleansNodeBaselines(t *testing.T) {