mirror of
https://github.com/MHSanaei/3x-ui.git
synced 2026-08-09 21:00:58 +00:00
fix(node): stop a departed master's frozen traffic from disabling clients (#6113)
client_global_traffics rows are keyed by (master_guid, email) and are only
ever overwritten by a push from that same master. A master that stops
pushing — decommissioned, reinstalled under a fresh GUID, or detached from
the node — therefore leaves its last snapshot behind permanently.
depletedClientsCond's cross-panel EXISTS branch matched any such row, so a
node kept comparing a client's quota against counters frozen weeks earlier.
Once they exceeded the quota the node disabled the client on every traffic
poll, and the node -> master enable merge latched that off on the master too,
where nothing sets it back. The reported symptom is exactly this: a client at
11 GB of a 24 GB quota, enabled on two nodes, disabled on the third, which
still held a 27-day-old row from a previous master reporting 30 GB.
Bound both the enforcement predicate and the display overlay to rows a master
refreshed within globalTrafficFreshWindow. Masters push every 30s, so a live
master is never affected; a master that is merely unreachable for a while
keeps enforcing for a full day before its numbers are set aside.
The one-way enable merge that makes such a disable permanent on the master is
deliberate (12d84c2a, #4917) and is left alone.
This commit is contained in:
@@ -429,8 +429,7 @@ func (s *ClientService) BulkAdjust(inboundSvc *InboundService, emails []string,
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
now := time.Now().Unix() * 1000
|
cond, condArgs := depletedCond(db)
|
||||||
cond := depletedCond(db)
|
|
||||||
candidateEmails := make([]string, 0, len(plan))
|
candidateEmails := make([]string, 0, len(plan))
|
||||||
for email, entry := range plan {
|
for email, entry := range plan {
|
||||||
if entry.applyExpiry || entry.applyTotal {
|
if entry.applyExpiry || entry.applyTotal {
|
||||||
@@ -441,7 +440,7 @@ func (s *ClientService) BulkAdjust(inboundSvc *InboundService, emails []string,
|
|||||||
for _, batch := range chunkStrings(candidateEmails, sqlInChunk) {
|
for _, batch := range chunkStrings(candidateEmails, sqlInChunk) {
|
||||||
var rows []string
|
var rows []string
|
||||||
if err := db.Model(xray.ClientTraffic{}).
|
if err := db.Model(xray.ClientTraffic{}).
|
||||||
Where(cond+" AND enable = ? AND email IN ?", now, false, batch).
|
Where(cond+" AND enable = ? AND email IN ?", append(append([]any{}, condArgs...), false, batch)...).
|
||||||
Pluck("email", &rows).Error; err != nil {
|
Pluck("email", &rows).Error; err != nil {
|
||||||
return result, needRestart, err
|
return result, needRestart, err
|
||||||
}
|
}
|
||||||
@@ -503,7 +502,7 @@ func (s *ClientService) BulkAdjust(inboundSvc *InboundService, emails []string,
|
|||||||
for _, batch := range chunkStrings(wasList, sqlInChunk) {
|
for _, batch := range chunkStrings(wasList, sqlInChunk) {
|
||||||
var rows []string
|
var rows []string
|
||||||
if err := db.Model(xray.ClientTraffic{}).
|
if err := db.Model(xray.ClientTraffic{}).
|
||||||
Where(cond+" AND email IN ?", now, batch).
|
Where(cond+" AND email IN ?", append(append([]any{}, condArgs...), batch)...).
|
||||||
Pluck("email", &rows).Error; err != nil {
|
Pluck("email", &rows).Error; err != nil {
|
||||||
return result, needRestart, err
|
return result, needRestart, err
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -2,6 +2,7 @@ package service
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"testing"
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
"github.com/mhsanaei/3x-ui/v3/internal/database"
|
"github.com/mhsanaei/3x-ui/v3/internal/database"
|
||||||
"github.com/mhsanaei/3x-ui/v3/internal/database/model"
|
"github.com/mhsanaei/3x-ui/v3/internal/database/model"
|
||||||
@@ -70,7 +71,7 @@ func TestDepletedCond_ProbeGuard(t *testing.T) {
|
|||||||
|
|
||||||
// No global rows: the cross-panel EXISTS branch is skipped (#5392), but a
|
// No global rows: the cross-panel EXISTS branch is skipped (#5392), but a
|
||||||
// client over its local quota is still disabled.
|
// client over its local quota is still disabled.
|
||||||
if got := depletedCond(db); got != depletedClientsCondLocal {
|
if got, _ := depletedCond(db); got != depletedClientsCondLocal {
|
||||||
t.Fatalf("empty globals must use the local-only predicate")
|
t.Fatalf("empty globals must use the local-only predicate")
|
||||||
}
|
}
|
||||||
seedClientRow(t, "local-cap", 1, 600, 600, 1000)
|
seedClientRow(t, "local-cap", 1, 600, 600, 1000)
|
||||||
@@ -85,11 +86,84 @@ func TestDepletedCond_ProbeGuard(t *testing.T) {
|
|||||||
if err := svc.AcceptGlobalTraffic("master-a", []*xray.ClientTraffic{{Email: "local-cap", Up: 1, Down: 1}}); err != nil {
|
if err := svc.AcceptGlobalTraffic("master-a", []*xray.ClientTraffic{{Email: "local-cap", Up: 1, Down: 1}}); err != nil {
|
||||||
t.Fatalf("AcceptGlobalTraffic: %v", err)
|
t.Fatalf("AcceptGlobalTraffic: %v", err)
|
||||||
}
|
}
|
||||||
if got := depletedCond(db); got != depletedClientsCond {
|
if got, _ := depletedCond(db); got != depletedClientsCond {
|
||||||
t.Fatalf("with globals present the cross-panel predicate must be used")
|
t.Fatalf("with globals present the cross-panel predicate must be used")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// A master that stopped pushing leaves its last snapshot behind forever. Those
|
||||||
|
// frozen counters must not keep enforcing quota, or a client well inside its
|
||||||
|
// limit is disabled on every traffic poll with no way back (#6113).
|
||||||
|
func TestStaleGlobalTraffic_Ignored(t *testing.T) {
|
||||||
|
db := initTrafficTestDB(t)
|
||||||
|
svc := &InboundService{}
|
||||||
|
|
||||||
|
staleAt := time.Now().Add(-globalTrafficFreshWindow - time.Hour).UnixMilli()
|
||||||
|
seedStaleGlobal := func(t *testing.T, guid, email string, up, down int64) {
|
||||||
|
t.Helper()
|
||||||
|
if err := svc.AcceptGlobalTraffic(guid, []*xray.ClientTraffic{{Email: email, Up: up, Down: down}}); err != nil {
|
||||||
|
t.Fatalf("AcceptGlobalTraffic(%s): %v", guid, err)
|
||||||
|
}
|
||||||
|
if err := db.Model(&model.ClientGlobalTraffic{}).
|
||||||
|
Where("master_guid = ? AND email = ?", guid, email).
|
||||||
|
Update("updated_at", staleAt).Error; err != nil {
|
||||||
|
t.Fatalf("age row: %v", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
t.Run("stale row alone neither enforces nor selects the cross-panel predicate", func(t *testing.T) {
|
||||||
|
// 200 of 1000 used locally, 1900 reported by a master gone for a day.
|
||||||
|
seedClientRow(t, "cap", 1, 100, 100, 1000)
|
||||||
|
seedStaleGlobal(t, "dead-master", "cap", 1000, 900)
|
||||||
|
|
||||||
|
if got, _ := depletedCond(db); got != depletedClientsCondLocal {
|
||||||
|
t.Fatalf("only stale globals must fall back to the local-only predicate")
|
||||||
|
}
|
||||||
|
if _, count, err := svc.disableInvalidClients(db); err != nil {
|
||||||
|
t.Fatalf("disableInvalidClients: %v", err)
|
||||||
|
} else if count != 0 {
|
||||||
|
t.Fatalf("stale global usage must not disable a client, disabled %d", count)
|
||||||
|
}
|
||||||
|
if got := readTraffic(t, db, "cap"); !got.Enable {
|
||||||
|
t.Error("client within its local quota must stay enabled")
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
|
t.Run("stale row does not inflate the displayed total", func(t *testing.T) {
|
||||||
|
rows := []*xray.ClientTraffic{{Email: "cap", Up: 100, Down: 100}}
|
||||||
|
overlayGlobalTraffic(db, rows)
|
||||||
|
if rows[0].Up != 100 || rows[0].Down != 100 {
|
||||||
|
t.Errorf("stale global must not overlay display counters, got up=%d down=%d", rows[0].Up, rows[0].Down)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
|
t.Run("a live master still enforces alongside the stale row", func(t *testing.T) {
|
||||||
|
// This is the #6113 shape: one dead master frozen over quota, one live
|
||||||
|
// master well under it. Only the live one may decide.
|
||||||
|
if err := svc.AcceptGlobalTraffic("live-master", []*xray.ClientTraffic{{Email: "cap", Up: 1, Down: 1}}); err != nil {
|
||||||
|
t.Fatalf("AcceptGlobalTraffic: %v", err)
|
||||||
|
}
|
||||||
|
if got, _ := depletedCond(db); got != depletedClientsCond {
|
||||||
|
t.Fatalf("a fresh global row must select the cross-panel predicate")
|
||||||
|
}
|
||||||
|
if _, count, err := svc.disableInvalidClients(db); err != nil {
|
||||||
|
t.Fatalf("disableInvalidClients: %v", err)
|
||||||
|
} else if count != 0 {
|
||||||
|
t.Fatalf("the live master reports usage well under quota, disabled %d", count)
|
||||||
|
}
|
||||||
|
|
||||||
|
// And once the live master reports real depletion, it takes effect.
|
||||||
|
if err := svc.AcceptGlobalTraffic("live-master", []*xray.ClientTraffic{{Email: "cap", Up: 600, Down: 500}}); err != nil {
|
||||||
|
t.Fatalf("AcceptGlobalTraffic: %v", err)
|
||||||
|
}
|
||||||
|
if _, count, err := svc.disableInvalidClients(db); err != nil {
|
||||||
|
t.Fatalf("disableInvalidClients: %v", err)
|
||||||
|
} else if count != 1 {
|
||||||
|
t.Fatalf("fresh cross-panel depletion must disable the client, disabled %d", count)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
func TestGlobalUsage_DisablesClient(t *testing.T) {
|
func TestGlobalUsage_DisablesClient(t *testing.T) {
|
||||||
db := initTrafficTestDB(t)
|
db := initTrafficTestDB(t)
|
||||||
svc := &InboundService{}
|
svc := &InboundService{}
|
||||||
|
|||||||
@@ -49,46 +49,67 @@ func (s *InboundService) disableInvalidInbounds(tx *gorm.DB) (bool, int64, error
|
|||||||
return needRestart, count, err
|
return needRestart, count, err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// globalTrafficFreshWindow bounds how long a pushed client_global_traffics row
|
||||||
|
// stays authoritative. Masters refresh their rows every nodeGlobalPushInterval
|
||||||
|
// (30s), so a row older than this belongs to a master that stopped pushing —
|
||||||
|
// decommissioned, reinstalled under a new GUID, or detached from this node.
|
||||||
|
// Such a row keeps its last-seen counters forever, and without this bound a
|
||||||
|
// long-dead master's numbers permanently trip the cross-panel quota check and
|
||||||
|
// disable clients that are nowhere near their limit (#6113). The window is far
|
||||||
|
// wider than any real push gap, so a master that is merely unreachable for a
|
||||||
|
// while keeps enforcing.
|
||||||
|
const globalTrafficFreshWindow = 24 * time.Hour
|
||||||
|
|
||||||
|
func globalTrafficFreshSince() int64 {
|
||||||
|
return time.Now().Add(-globalTrafficFreshWindow).UnixMilli()
|
||||||
|
}
|
||||||
|
|
||||||
// depletedClientsCond matches clients that exhausted their quota or expired.
|
// depletedClientsCond matches clients that exhausted their quota or expired.
|
||||||
// Besides the local counters it also trips on the cross-panel usage a master
|
// Besides the local counters it also trips on the cross-panel usage a master
|
||||||
// pushed into client_global_traffics — that's what lets a node cut a client
|
// pushed into client_global_traffics — that's what lets a node cut a client
|
||||||
// whose combined usage exceeds the quota even though the local share doesn't
|
// whose combined usage exceeds the quota even though the local share doesn't.
|
||||||
// (placeholders: now).
|
// Only rows a master refreshed recently count (placeholders: now, freshSince).
|
||||||
const depletedClientsCond = `((total > 0 AND up + down >= total)
|
const depletedClientsCond = `((total > 0 AND up + down >= total)
|
||||||
OR (expiry_time > 0 AND expiry_time <= ?)
|
OR (expiry_time > 0 AND expiry_time <= ?)
|
||||||
OR (total > 0 AND EXISTS (
|
OR (total > 0 AND EXISTS (
|
||||||
SELECT 1 FROM client_global_traffics g
|
SELECT 1 FROM client_global_traffics g
|
||||||
WHERE g.email = client_traffics.email AND g.up + g.down >= client_traffics.total
|
WHERE g.email = client_traffics.email
|
||||||
|
AND g.updated_at >= ?
|
||||||
|
AND g.up + g.down >= client_traffics.total
|
||||||
)))`
|
)))`
|
||||||
|
|
||||||
// depletedClientsCondLocal is depletedClientsCond without the cross-panel
|
// depletedClientsCondLocal is depletedClientsCond without the cross-panel
|
||||||
// client_global_traffics check. The EXISTS branch is a correlated subquery that
|
// client_global_traffics check. The EXISTS branch is a correlated subquery that
|
||||||
// turns every traffic poll into a full client_traffics scan; on a panel no
|
// turns every traffic poll into a full client_traffics scan; on a panel no
|
||||||
// master pushes to (the common case) client_global_traffics is empty, so the
|
// master pushes to (the common case) client_global_traffics is empty, so the
|
||||||
// branch can never match and is pure CPU cost (#5392).
|
// branch can never match and is pure CPU cost (#5392). Placeholders: now.
|
||||||
const depletedClientsCondLocal = `((total > 0 AND up + down >= total)
|
const depletedClientsCondLocal = `((total > 0 AND up + down >= total)
|
||||||
OR (expiry_time > 0 AND expiry_time <= ?))`
|
OR (expiry_time > 0 AND expiry_time <= ?))`
|
||||||
|
|
||||||
// depletedCond returns the local-only predicate unless this panel actually
|
// depletedCond returns the predicate matching depleted clients together with
|
||||||
// holds global-traffic rows, in which case the cross-panel EXISTS check is
|
// the arguments it binds. The local-only variant is used unless this panel
|
||||||
// needed to enforce combined quota. Both variants take the same single
|
// holds a global-traffic row a master still refreshes, in which case the
|
||||||
// expiry_time placeholder, so callers pass identical args either way.
|
// cross-panel EXISTS check is needed to enforce combined quota.
|
||||||
func depletedCond(tx *gorm.DB) string {
|
func depletedCond(tx *gorm.DB) (string, []any) {
|
||||||
|
now := time.Now().UnixMilli()
|
||||||
|
freshSince := globalTrafficFreshSince()
|
||||||
var probe int64
|
var probe int64
|
||||||
if err := tx.Model(&model.ClientGlobalTraffic{}).Limit(1).Count(&probe).Error; err == nil && probe > 0 {
|
err := tx.Model(&model.ClientGlobalTraffic{}).
|
||||||
return depletedClientsCond
|
Where("updated_at >= ?", freshSince).
|
||||||
|
Limit(1).Count(&probe).Error
|
||||||
|
if err == nil && probe > 0 {
|
||||||
|
return depletedClientsCond, []any{now, freshSince}
|
||||||
}
|
}
|
||||||
return depletedClientsCondLocal
|
return depletedClientsCondLocal, []any{now}
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *InboundService) disableInvalidClients(tx *gorm.DB) (bool, int64, error) {
|
func (s *InboundService) disableInvalidClients(tx *gorm.DB) (bool, int64, error) {
|
||||||
now := time.Now().Unix() * 1000
|
|
||||||
needRestart := false
|
needRestart := false
|
||||||
cond := depletedCond(tx)
|
cond, condArgs := depletedCond(tx)
|
||||||
|
|
||||||
var depletedRows []xray.ClientTraffic
|
var depletedRows []xray.ClientTraffic
|
||||||
err := tx.Model(xray.ClientTraffic{}).
|
err := tx.Model(xray.ClientTraffic{}).
|
||||||
Where(cond+" AND enable = ?", now, true).
|
Where(cond+" AND enable = ?", append(condArgs, true)...).
|
||||||
Find(&depletedRows).Error
|
Find(&depletedRows).Error
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return false, 0, err
|
return false, 0, err
|
||||||
@@ -185,7 +206,7 @@ func (s *InboundService) disableInvalidClients(tx *gorm.DB) (bool, int64, error)
|
|||||||
if len(depletedEmails) > 0 {
|
if len(depletedEmails) > 0 {
|
||||||
if err := tx.Model(&model.ClientRecord{}).
|
if err := tx.Model(&model.ClientRecord{}).
|
||||||
Where("email IN ?", depletedEmails).
|
Where("email IN ?", depletedEmails).
|
||||||
Updates(map[string]any{"enable": false, "updated_at": now}).Error; err != nil {
|
Updates(map[string]any{"enable": false, "updated_at": time.Now().UnixMilli()}).Error; err != nil {
|
||||||
logger.Warning("disableInvalidClients update clients.enable:", err)
|
logger.Warning("disableInvalidClients update clients.enable:", err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -100,15 +100,21 @@ func chunkGlobalRows(rows []model.ClientGlobalTraffic, size int) [][]model.Clien
|
|||||||
}
|
}
|
||||||
|
|
||||||
// overlayGlobalTraffic raises Up/Down on the given rows to the largest global
|
// overlayGlobalTraffic raises Up/Down on the given rows to the largest global
|
||||||
// value any master pushed for that email. Read-path only — callers hand it
|
// value any master still pushing for that email reported. Read-path only —
|
||||||
// rows about to be serialized for display; the stored counters are untouched.
|
// 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).
|
||||||
func overlayGlobalTraffic(db *gorm.DB, rows []*xray.ClientTraffic) {
|
func overlayGlobalTraffic(db *gorm.DB, rows []*xray.ClientTraffic) {
|
||||||
if len(rows) == 0 {
|
if len(rows) == 0 {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
freshSince := globalTrafficFreshSince()
|
||||||
// Cheap short-circuit for the common case (a panel no master pushes to).
|
// Cheap short-circuit for the common case (a panel no master pushes to).
|
||||||
var probe int64
|
var probe int64
|
||||||
if err := db.Model(&model.ClientGlobalTraffic{}).Limit(1).Count(&probe).Error; err != nil || probe == 0 {
|
if err := db.Model(&model.ClientGlobalTraffic{}).
|
||||||
|
Where("updated_at >= ?", freshSince).
|
||||||
|
Limit(1).Count(&probe).Error; err != nil || probe == 0 {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -126,7 +132,7 @@ func overlayGlobalTraffic(db *gorm.DB, rows []*xray.ClientTraffic) {
|
|||||||
}
|
}
|
||||||
for _, batch := range chunkStrings(emails, sqlInChunk) {
|
for _, batch := range chunkStrings(emails, sqlInChunk) {
|
||||||
var globals []model.ClientGlobalTraffic
|
var globals []model.ClientGlobalTraffic
|
||||||
if err := db.Where("email IN ?", batch).Find(&globals).Error; err != nil {
|
if err := db.Where("email IN ? AND updated_at >= ?", batch, freshSince).Find(&globals).Error; err != nil {
|
||||||
logger.Warning("overlayGlobalTraffic:", err)
|
logger.Warning("overlayGlobalTraffic:", err)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user