mirror of
https://github.com/MHSanaei/3x-ui.git
synced 2026-11-02 03:05:02 +03:00
fix(traffic): keep lifecycle statements under SQLite's variable limit
Three statement shapes bound one SQL variable per row and failed past
SQLite's limit, which a node can reach by itself with the clients and
inbounds it reports: the depleted-client lookup and disable in
disableInvalidClients (stopping every expiry and quota check), the
whole-slice Saves of renewed inbounds and traffics (stopping renewals),
and Preload("ClientStats") over all inbounds (stopping config builds).
The IN lists are chunked, the Saves go 100 rows at a time, and client
stats are attached in chunks instead of preloaded.
This commit is contained in:
@@ -248,3 +248,77 @@ func TestAcceptGlobalTrafficDropsAnOversizedMasterGuid(t *testing.T) {
|
||||
t.Fatalf("stored %d row(s) under an oversized master GUID", rows)
|
||||
}
|
||||
}
|
||||
|
||||
// A node can report tens of thousands of its own clients over quota; one IN list with a
|
||||
// bound variable per email overran SQLite's limit and stopped every expiry and quota check.
|
||||
func TestDisableInvalidClientsPastTheSQLVariableLimit(t *testing.T) {
|
||||
db := initTrafficTestDB(t)
|
||||
const n = 33000
|
||||
if err := db.Create(&model.Inbound{UserId: 1, Tag: "big", Port: 41100, Protocol: model.VLESS, Enable: true, Settings: `{"clients":[]}`}).Error; err != nil {
|
||||
t.Fatalf("seed inbound: %v", err)
|
||||
}
|
||||
for _, stmt := range []string{
|
||||
`WITH RECURSIVE seq(i) AS (SELECT 1 UNION ALL SELECT i+1 FROM seq WHERE i < ?)
|
||||
INSERT INTO client_traffics (inbound_id, email, enable, total, up, down) SELECT 1, 'c' || i, true, 1, 1, 1 FROM seq`,
|
||||
`WITH RECURSIVE seq(i) AS (SELECT 1 UNION ALL SELECT i+1 FROM seq WHERE i < ?)
|
||||
INSERT INTO clients (email, enable, sub_id) SELECT 'c' || i, true, 's' || i FROM seq`,
|
||||
} {
|
||||
if err := db.Exec(stmt, n).Error; err != nil {
|
||||
t.Fatalf("seed: %v", err)
|
||||
}
|
||||
}
|
||||
if err := db.Exec(`INSERT INTO client_inbounds (client_id, inbound_id) SELECT id, 1 FROM clients`).Error; err != nil {
|
||||
t.Fatalf("link clients: %v", err)
|
||||
}
|
||||
_, count, _, err := (&InboundService{}).disableInvalidClients(db, newTrafficMutationBatch())
|
||||
if err != nil || count != n {
|
||||
t.Fatalf("disableInvalidClients = %d, %v; want all %d disabled", count, err, n)
|
||||
}
|
||||
}
|
||||
|
||||
// Preloading client stats binds one variable per inbound, so past SQLite's limit the
|
||||
// master could not even build Xray's config; nodes report inbounds into the same table.
|
||||
func TestGetAllInboundsPastTheSQLVariableLimit(t *testing.T) {
|
||||
db := initTrafficTestDB(t)
|
||||
const n = 33000
|
||||
if err := db.Exec(`WITH RECURSIVE seq(i) AS (SELECT 1 UNION ALL SELECT i+1 FROM seq WHERE i < ?)
|
||||
INSERT INTO inbounds (user_id, tag, port, protocol, enable, settings, stream_settings, sniffing, listen, remark)
|
||||
SELECT 1, 'in-' || i, i, 'vless', true, '{}', '', '', '', '' FROM seq`, n).Error; err != nil {
|
||||
t.Fatalf("seed inbounds: %v", err)
|
||||
}
|
||||
seedClientRow(t, "alice", n, 0, 0, 0)
|
||||
inbounds, err := (&InboundService{}).GetAllInbounds()
|
||||
if err != nil || len(inbounds) != n {
|
||||
t.Fatalf("GetAllInbounds = %d inbounds, %v; want %d", len(inbounds), err, n)
|
||||
}
|
||||
if last := inbounds[len(inbounds)-1]; len(last.ClientStats) != 1 || last.ClientStats[0].Email != "alice" {
|
||||
t.Fatalf("last inbound's client stats = %v, want alice's row", last.ClientStats)
|
||||
}
|
||||
}
|
||||
|
||||
// Saving every renewed inbound in one statement binds every column of every row, which
|
||||
// overran SQLite's variable limit once enough inbounds held a client due for renewal.
|
||||
func TestAutoRenewClientsPastTheSQLVariableLimit(t *testing.T) {
|
||||
db := initTrafficTestDB(t)
|
||||
const n = 1400
|
||||
for _, stmt := range []string{
|
||||
`WITH RECURSIVE seq(i) AS (SELECT 1 UNION ALL SELECT i+1 FROM seq WHERE i < ?)
|
||||
INSERT INTO inbounds (user_id, tag, port, protocol, enable, settings, stream_settings, sniffing, listen, remark)
|
||||
SELECT 1, 'in-' || i, i, 'vless', true, '{"clients":[{"email":"c' || i || '","enable":true,"expiryTime":1000,"reset":30}]}', '', '', '', '' FROM seq`,
|
||||
`WITH RECURSIVE seq(i) AS (SELECT 1 UNION ALL SELECT i+1 FROM seq WHERE i < ?)
|
||||
INSERT INTO clients (email, enable, sub_id, expiry_time, reset) SELECT 'c' || i, true, 's' || i, 1000, 30 FROM seq`,
|
||||
`WITH RECURSIVE seq(i) AS (SELECT 1 UNION ALL SELECT i+1 FROM seq WHERE i < ?)
|
||||
INSERT INTO client_traffics (inbound_id, email, enable, expiry_time, reset) SELECT i, 'c' || i, true, 1000, 30 FROM seq`,
|
||||
} {
|
||||
if err := db.Exec(stmt, n).Error; err != nil {
|
||||
t.Fatalf("seed: %v", err)
|
||||
}
|
||||
}
|
||||
if err := db.Exec(`INSERT INTO client_inbounds (client_id, inbound_id) SELECT c.id, i.id FROM clients c JOIN inbounds i ON i.tag = 'in-' || substr(c.email, 2)`).Error; err != nil {
|
||||
t.Fatalf("link clients: %v", err)
|
||||
}
|
||||
_, count, err := (&InboundService{}).autoRenewClients(db, newTrafficMutationBatch())
|
||||
if err != nil || count != n {
|
||||
t.Fatalf("autoRenewClients = %d, %v; want all %d renewed", count, err, n)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -192,7 +192,10 @@ func normalizeInboundShareAddressColumns(tx *gorm.DB) error {
|
||||
func (s *InboundService) GetInbounds(userId int) ([]*model.Inbound, error) {
|
||||
db := database.GetDB()
|
||||
var inbounds []*model.Inbound
|
||||
err := db.Model(model.Inbound{}).Preload("ClientStats").Where("user_id = ?", userId).Order("id ASC").Find(&inbounds).Error
|
||||
err := db.Model(model.Inbound{}).Where("user_id = ?", userId).Order("id ASC").Find(&inbounds).Error
|
||||
if err == nil {
|
||||
err = attachClientStats(db, inbounds)
|
||||
}
|
||||
if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return nil, err
|
||||
}
|
||||
@@ -235,7 +238,10 @@ func (s *InboundService) annotateLocalOriginGuid(inbounds []*model.Inbound) {
|
||||
func (s *InboundService) GetInboundsSlim(userId int) ([]*model.Inbound, error) {
|
||||
db := database.GetDB()
|
||||
var inbounds []*model.Inbound
|
||||
err := db.Model(model.Inbound{}).Preload("ClientStats").Where("user_id = ?", userId).Order("id ASC").Find(&inbounds).Error
|
||||
err := db.Model(model.Inbound{}).Where("user_id = ?", userId).Order("id ASC").Find(&inbounds).Error
|
||||
if err == nil {
|
||||
err = attachClientStats(db, inbounds)
|
||||
}
|
||||
if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return nil, err
|
||||
}
|
||||
@@ -552,7 +558,10 @@ func inboundMtprotoDomain(protocol string, settings string) string {
|
||||
func (s *InboundService) GetAllInbounds() ([]*model.Inbound, error) {
|
||||
db := database.GetDB()
|
||||
var inbounds []*model.Inbound
|
||||
err := db.Model(model.Inbound{}).Preload("ClientStats").Find(&inbounds).Error
|
||||
err := db.Model(model.Inbound{}).Find(&inbounds).Error
|
||||
if err == nil {
|
||||
err = attachClientStats(db, inbounds)
|
||||
}
|
||||
if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return nil, err
|
||||
}
|
||||
@@ -2620,7 +2629,10 @@ func (s *InboundService) GetClientReverseTags() (string, error) {
|
||||
func (s *InboundService) SearchInbounds(query string) ([]*model.Inbound, error) {
|
||||
db := database.GetDB()
|
||||
var inbounds []*model.Inbound
|
||||
err := db.Model(model.Inbound{}).Preload("ClientStats").Where("remark like ?", "%"+query+"%").Find(&inbounds).Error
|
||||
err := db.Model(model.Inbound{}).Where("remark like ?", "%"+query+"%").Find(&inbounds).Error
|
||||
if err == nil {
|
||||
err = attachClientStats(db, inbounds)
|
||||
}
|
||||
if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
@@ -105,7 +105,10 @@ func (s *InboundService) disableInvalidClients(tx *gorm.DB, mutationBatch *traff
|
||||
Email string
|
||||
}
|
||||
var targets []target
|
||||
if len(depletedEmails) > 0 {
|
||||
// Chunked like every other statement here: a node alone can report more depleted
|
||||
// clients than SQLite binds variables, which would stop the whole tick.
|
||||
for _, batch := range chunkStrings(depletedEmails, sqlInChunk) {
|
||||
var found []target
|
||||
err = tx.Raw(`
|
||||
SELECT inbounds.id AS inbound_id, inbounds.node_id AS node_id,
|
||||
inbounds.tag AS tag, clients.email AS email
|
||||
@@ -113,10 +116,11 @@ func (s *InboundService) disableInvalidClients(tx *gorm.DB, mutationBatch *traff
|
||||
JOIN client_inbounds ON client_inbounds.client_id = clients.id
|
||||
JOIN inbounds ON inbounds.id = client_inbounds.inbound_id
|
||||
WHERE clients.email IN ?
|
||||
`, depletedEmails).Scan(&targets).Error
|
||||
`, batch).Scan(&found).Error
|
||||
if err != nil {
|
||||
return false, 0, nil, err
|
||||
}
|
||||
targets = append(targets, found...)
|
||||
}
|
||||
|
||||
byInbound := make(map[int][]target)
|
||||
@@ -167,9 +171,9 @@ func (s *InboundService) disableInvalidClients(tx *gorm.DB, mutationBatch *traff
|
||||
count += result.RowsAffected
|
||||
}
|
||||
|
||||
if len(depletedEmails) > 0 {
|
||||
for _, batch := range chunkStrings(depletedEmails, sqlInChunk) {
|
||||
if err := tx.Model(&model.ClientRecord{}).
|
||||
Where("email IN ?", depletedEmails).
|
||||
Where("email IN ?", batch).
|
||||
Updates(map[string]any{"enable": false, "updated_at": now}).Error; err != nil {
|
||||
return false, count, nil, err
|
||||
}
|
||||
|
||||
@@ -196,7 +196,7 @@ func (s *InboundService) MigrationRequirements() (err error) {
|
||||
return
|
||||
}
|
||||
}
|
||||
if err = tx.Save(inbounds).Error; err != nil {
|
||||
if err = saveInChunks(tx, inbounds); err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
|
||||
@@ -335,7 +335,7 @@ func (s *InboundService) adjustTraffics(tx *gorm.DB, dbClientTraffics []*xray.Cl
|
||||
}
|
||||
}
|
||||
|
||||
err = tx.Save(inbounds).Error
|
||||
err = saveInChunks(tx, inbounds)
|
||||
if err != nil {
|
||||
logger.Warning("AddClientTraffic update inbounds ", err)
|
||||
logger.Error(inbounds)
|
||||
@@ -533,7 +533,7 @@ func (s *InboundService) autoRenewClients(tx *gorm.DB, mutationBatch *trafficMut
|
||||
}
|
||||
inbounds[inbound_index].Settings = string(newSettings)
|
||||
}
|
||||
err = tx.Save(inbounds).Error
|
||||
err = saveInChunks(tx, inbounds)
|
||||
if err != nil {
|
||||
return false, 0, err
|
||||
}
|
||||
@@ -550,7 +550,7 @@ func (s *InboundService) autoRenewClients(tx *gorm.DB, mutationBatch *trafficMut
|
||||
logger.Warning("autoRenewClients sync clients: SyncInbound failed", syncErr)
|
||||
}
|
||||
}
|
||||
err = tx.Save(traffics).Error
|
||||
err = saveInChunks(tx, traffics)
|
||||
if err != nil {
|
||||
return false, 0, err
|
||||
}
|
||||
|
||||
@@ -1,6 +1,13 @@
|
||||
package service
|
||||
|
||||
import "strings"
|
||||
import (
|
||||
"strings"
|
||||
|
||||
"gorm.io/gorm"
|
||||
|
||||
"github.com/mhsanaei/3x-ui/v3/internal/database/model"
|
||||
"github.com/mhsanaei/3x-ui/v3/internal/xray"
|
||||
)
|
||||
|
||||
// sqliteMaxVars is a safe ceiling for the number of bind parameters in a
|
||||
// single SQL statement. SQLite's SQLITE_MAX_VARIABLE_NUMBER is 999 on builds
|
||||
@@ -102,3 +109,38 @@ func chunkCount(n, size int) int {
|
||||
}
|
||||
return count
|
||||
}
|
||||
|
||||
// attachClientStats fills each inbound's ClientStats in chunks: GORM's Preload binds one
|
||||
// variable per inbound, and nodes alone can report more inbounds than SQLite binds.
|
||||
func attachClientStats(db *gorm.DB, inbounds []*model.Inbound) error {
|
||||
byID := make(map[int]*model.Inbound, len(inbounds))
|
||||
ids := make([]int, 0, len(inbounds))
|
||||
for _, ib := range inbounds {
|
||||
byID[ib.Id] = ib
|
||||
ids = append(ids, ib.Id)
|
||||
}
|
||||
for _, batch := range chunkInts(ids, sqliteMaxVars) {
|
||||
var stats []xray.ClientTraffic
|
||||
if err := db.Where("inbound_id IN ?", batch).Order("id ASC").Find(&stats).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
for _, st := range stats {
|
||||
if ib := byID[st.InboundId]; ib != nil {
|
||||
ib.ClientStats = append(ib.ClientStats, st)
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// saveInChunks saves rows a slice at a time: one slice Save binds every column of every
|
||||
// row in a single statement, past SQLite's variable limit once there are enough rows.
|
||||
func saveInChunks[T any](tx *gorm.DB, rows []T) error {
|
||||
const rowsPerStatement = 100
|
||||
for start := 0; start < len(rows); start += rowsPerStatement {
|
||||
if err := tx.Save(rows[start:min(start+rowsPerStatement, len(rows))]).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user