diff --git a/internal/web/service/global_traffic_test.go b/internal/web/service/global_traffic_test.go index 83c935aee..8412f0768 100644 --- a/internal/web/service/global_traffic_test.go +++ b/internal/web/service/global_traffic_test.go @@ -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) + } +} diff --git a/internal/web/service/inbound.go b/internal/web/service/inbound.go index 27d10dbee..24c67327f 100644 --- a/internal/web/service/inbound.go +++ b/internal/web/service/inbound.go @@ -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 } diff --git a/internal/web/service/inbound_disable.go b/internal/web/service/inbound_disable.go index 72264a422..c29b68fda 100644 --- a/internal/web/service/inbound_disable.go +++ b/internal/web/service/inbound_disable.go @@ -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 } diff --git a/internal/web/service/inbound_migration.go b/internal/web/service/inbound_migration.go index 3030c4531..e3e96e192 100644 --- a/internal/web/service/inbound_migration.go +++ b/internal/web/service/inbound_migration.go @@ -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 } diff --git a/internal/web/service/inbound_traffic.go b/internal/web/service/inbound_traffic.go index fb8129377..6456aa1a9 100644 --- a/internal/web/service/inbound_traffic.go +++ b/internal/web/service/inbound_traffic.go @@ -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 } diff --git a/internal/web/service/inbound_util.go b/internal/web/service/inbound_util.go index 2917612d0..3f2d5c22d 100644 --- a/internal/web/service/inbound_util.go +++ b/internal/web/service/inbound_util.go @@ -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 +}