mirror of
https://github.com/MHSanaei/3x-ui.git
synced 2026-11-01 10:45:00 +03:00
fix(node-sync): keep shared-email lookups on the open transaction
emailUsedByOtherInbounds and emailsUsedByOtherInbounds always queried through database.GetDB(), yet five callers run them inside a transaction: both node traffic sync sweeps (setRemoteTrafficLocked), the inbound edit stats sync (updateClientTraffics) and both email-drop branches of UpdateInboundClient. Each call needed a second pool connection while the transaction held row locks on client_traffics/node_client_traffics. On PostgreSQL, once subscription fetches (UPDATE client_traffics SET last_sub_fetch) blocked on those locks filled the rest of the 25-connection pool, the lookup waited forever, the transaction never ended, and the panel hung until restart. Both helpers now take the *gorm.DB to query on; transactional callers pass tx and the rest pass the pool handle they already hold. The new tests cap the pool at one connection, so any query that escapes the transaction blocks; each went red before the fix on SQLite and (node sync) on a live PostgreSQL 16, and passes after. Closes #6789
This commit is contained in:
@@ -1143,7 +1143,7 @@ func (s *ClientService) bulkDelInboundClients(
|
|||||||
var sharedSet map[string]bool
|
var sharedSet map[string]bool
|
||||||
if !keepTraffic {
|
if !keepTraffic {
|
||||||
var sharedErr error
|
var sharedErr error
|
||||||
sharedSet, sharedErr = inboundSvc.emailsUsedByOtherInbounds(foundList, inboundId)
|
sharedSet, sharedErr = inboundSvc.emailsUsedByOtherInbounds(db, foundList, inboundId)
|
||||||
if sharedErr != nil {
|
if sharedErr != nil {
|
||||||
for email := range wantedEmails {
|
for email := range wantedEmails {
|
||||||
res.perEmailSkipped[email] = sharedErr.Error()
|
res.perEmailSkipped[email] = sharedErr.Error()
|
||||||
|
|||||||
@@ -92,14 +92,14 @@ func TestEmailsUsedByOtherInboundsMatchesCaseInsensitively(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
shared, err := is.emailsUsedByOtherInbounds([]string{"alice@x"}, ibA.Id)
|
shared, err := is.emailsUsedByOtherInbounds(database.GetDB(), []string{"alice@x"}, ibA.Id)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("emailsUsedByOtherInbounds: %v", err)
|
t.Fatalf("emailsUsedByOtherInbounds: %v", err)
|
||||||
}
|
}
|
||||||
if !shared["alice@x"] {
|
if !shared["alice@x"] {
|
||||||
t.Error("lower-cased lookup missed a stored mixed-case email")
|
t.Error("lower-cased lookup missed a stored mixed-case email")
|
||||||
}
|
}
|
||||||
used, err := is.emailUsedByOtherInbounds("alice@x", ibA.Id)
|
used, err := is.emailUsedByOtherInbounds(database.GetDB(), "alice@x", ibA.Id)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("emailUsedByOtherInbounds: %v", err)
|
t.Fatalf("emailUsedByOtherInbounds: %v", err)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -136,7 +136,7 @@ func (s *ClientService) delInboundClients(inboundSvc *InboundService, inboundId
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
var sharedErr error
|
var sharedErr error
|
||||||
sharedSet, sharedErr = inboundSvc.emailsUsedByOtherInbounds(removedEmails, inboundId)
|
sharedSet, sharedErr = inboundSvc.emailsUsedByOtherInbounds(db, removedEmails, inboundId)
|
||||||
if sharedErr != nil {
|
if sharedErr != nil {
|
||||||
return false, sharedErr
|
return false, sharedErr
|
||||||
}
|
}
|
||||||
@@ -946,7 +946,7 @@ func (s *ClientService) UpdateInboundClient(inboundSvc *InboundService, data *mo
|
|||||||
return e
|
return e
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
stillUsed, sErr := inboundSvc.emailUsedByOtherInbounds(oldEmail, data.Id)
|
stillUsed, sErr := inboundSvc.emailUsedByOtherInbounds(tx, oldEmail, data.Id)
|
||||||
if sErr != nil {
|
if sErr != nil {
|
||||||
return sErr
|
return sErr
|
||||||
}
|
}
|
||||||
@@ -968,7 +968,7 @@ func (s *ClientService) UpdateInboundClient(inboundSvc *InboundService, data *mo
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
stillUsed, sErr := inboundSvc.emailUsedByOtherInbounds(oldEmail, data.Id)
|
stillUsed, sErr := inboundSvc.emailUsedByOtherInbounds(tx, oldEmail, data.Id)
|
||||||
if sErr != nil {
|
if sErr != nil {
|
||||||
return sErr
|
return sErr
|
||||||
}
|
}
|
||||||
@@ -1140,7 +1140,7 @@ func (s *ClientService) DelInboundClientByEmail(inboundSvc *InboundService, inbo
|
|||||||
prevSettings := oldInbound.Settings
|
prevSettings := oldInbound.Settings
|
||||||
oldInbound.Settings = string(newSettings)
|
oldInbound.Settings = string(newSettings)
|
||||||
|
|
||||||
emailShared, err := inboundSvc.emailUsedByOtherInbounds(email, inboundId)
|
emailShared, err := inboundSvc.emailUsedByOtherInbounds(db, email, inboundId)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return false, err
|
return false, err
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -2290,7 +2290,7 @@ func (s *InboundService) updateClientTraffics(tx *gorm.DB, oldInbound *model.Inb
|
|||||||
if _, kept := newEmails[email]; kept {
|
if _, kept := newEmails[email]; kept {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
stillUsed, err := s.emailUsedByOtherInbounds(email, oldInbound.Id)
|
stillUsed, err := s.emailUsedByOtherInbounds(tx, email, oldInbound.Id)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -124,14 +124,14 @@ func (s *InboundService) backfillClientStats(db *gorm.DB, inbounds []*model.Inbo
|
|||||||
return clientsByInbound
|
return clientsByInbound
|
||||||
}
|
}
|
||||||
|
|
||||||
// emailUsedByOtherInbounds reports whether email lives in any inbound other
|
// emailUsedByOtherInbounds reports whether email lives in any inbound other than
|
||||||
// than exceptInboundId. Empty email returns false.
|
// exceptInboundId. Inside a transaction pass tx: a pool query there starves the pool.
|
||||||
func (s *InboundService) emailUsedByOtherInbounds(email string, exceptInboundId int) (bool, error) {
|
func (s *InboundService) emailUsedByOtherInbounds(db *gorm.DB, email string, exceptInboundId int) (bool, error) {
|
||||||
if email == "" {
|
if email == "" {
|
||||||
return false, nil
|
return false, nil
|
||||||
}
|
}
|
||||||
var count int64
|
var count int64
|
||||||
err := database.GetDB().Table("client_inbounds").
|
err := db.Table("client_inbounds").
|
||||||
Joins("JOIN clients ON clients.id = client_inbounds.client_id").
|
Joins("JOIN clients ON clients.id = client_inbounds.client_id").
|
||||||
Where("client_inbounds.inbound_id != ? AND LOWER(clients.email) = ?",
|
Where("client_inbounds.inbound_id != ? AND LOWER(clients.email) = ?",
|
||||||
exceptInboundId, strings.ToLower(strings.TrimSpace(email))).
|
exceptInboundId, strings.ToLower(strings.TrimSpace(email))).
|
||||||
@@ -142,7 +142,7 @@ func (s *InboundService) emailUsedByOtherInbounds(email string, exceptInboundId
|
|||||||
return count > 0, nil
|
return count > 0, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *InboundService) emailsUsedByOtherInbounds(emails []string, exceptInboundId int) (map[string]bool, error) {
|
func (s *InboundService) emailsUsedByOtherInbounds(db *gorm.DB, emails []string, exceptInboundId int) (map[string]bool, error) {
|
||||||
shared := make(map[string]bool, len(emails))
|
shared := make(map[string]bool, len(emails))
|
||||||
want := make(map[string]struct{}, len(emails))
|
want := make(map[string]struct{}, len(emails))
|
||||||
for _, e := range emails {
|
for _, e := range emails {
|
||||||
@@ -158,7 +158,6 @@ func (s *InboundService) emailsUsedByOtherInbounds(emails []string, exceptInboun
|
|||||||
for e := range want {
|
for e := range want {
|
||||||
lowered = append(lowered, e)
|
lowered = append(lowered, e)
|
||||||
}
|
}
|
||||||
db := database.GetDB()
|
|
||||||
for _, batch := range chunkStrings(lowered, sqlInChunk) {
|
for _, batch := range chunkStrings(lowered, sqlInChunk) {
|
||||||
var rows []struct{ Email string }
|
var rows []struct{ Email string }
|
||||||
err := db.Table("client_inbounds").
|
err := db.Table("client_inbounds").
|
||||||
|
|||||||
@@ -940,7 +940,7 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi
|
|||||||
// email's last inbound — wiping it while a sibling still feeds it
|
// email's last inbound — wiping it while a sibling still feeds it
|
||||||
// loses the summed history, and the next node sync would re-seed the
|
// loses the summed history, and the next node sync would re-seed the
|
||||||
// row with that node's counter alone.
|
// row with that node's counter alone.
|
||||||
sharedEmails, sErr := s.emailsUsedByOtherInbounds(goneEmails, c.Id)
|
sharedEmails, sErr := s.emailsUsedByOtherInbounds(tx, goneEmails, c.Id)
|
||||||
if sErr != nil {
|
if sErr != nil {
|
||||||
return false, sErr
|
return false, sErr
|
||||||
}
|
}
|
||||||
@@ -1218,7 +1218,7 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi
|
|||||||
}
|
}
|
||||||
// Same shared-accumulator rule as the inbound-removal sweep above:
|
// Same shared-accumulator rule as the inbound-removal sweep above:
|
||||||
// keep the row while another inbound still references the email.
|
// keep the row while another inbound still references the email.
|
||||||
stillUsed, uErr := s.emailUsedByOtherInbounds(existing.Email, c.Id)
|
stillUsed, uErr := s.emailUsedByOtherInbounds(tx, existing.Email, c.Id)
|
||||||
if uErr != nil {
|
if uErr != nil {
|
||||||
return false, uErr
|
return false, uErr
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,180 @@
|
|||||||
|
package service
|
||||||
|
|
||||||
|
import (
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"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"
|
||||||
|
)
|
||||||
|
|
||||||
|
// capPoolAtOneConn models a pool whose other connections are all parked on the
|
||||||
|
// transaction's row locks: any query outside tx then waits forever.
|
||||||
|
func capPoolAtOneConn(t *testing.T) {
|
||||||
|
t.Helper()
|
||||||
|
sqlDB, err := database.GetDB().DB()
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("sql.DB: %v", err)
|
||||||
|
}
|
||||||
|
sqlDB.SetMaxOpenConns(1)
|
||||||
|
}
|
||||||
|
|
||||||
|
func runWithoutSecondConn(t *testing.T, name string, fn func() error) error {
|
||||||
|
t.Helper()
|
||||||
|
done := make(chan error, 1)
|
||||||
|
go func() { done <- fn() }()
|
||||||
|
select {
|
||||||
|
case err := <-done:
|
||||||
|
return err
|
||||||
|
case <-time.After(5 * time.Second):
|
||||||
|
// The DB cleanup's Close fails the parked pool request, which unwinds fn.
|
||||||
|
t.Fatalf("%s blocked waiting for a second pool connection while its transaction is open", name)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func createNodeInboundWithSettings(t *testing.T, db *gorm.DB, nodeID int, tag string, port int, settings string) {
|
||||||
|
t.Helper()
|
||||||
|
nid := nodeID
|
||||||
|
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 syncNodeSnapshot(t *testing.T, svc *InboundService, nodeID int, inbounds ...*model.Inbound) error {
|
||||||
|
t.Helper()
|
||||||
|
_, err := svc.setRemoteTrafficLocked(nodeID, &runtime.TrafficSnapshot{Inbounds: inbounds}, false, false)
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
func assertNoTraffic(t *testing.T, email string) {
|
||||||
|
t.Helper()
|
||||||
|
var n int64
|
||||||
|
if err := database.GetDB().Model(xray.ClientTraffic{}).Where("email = ?", email).Count(&n).Error; err != nil {
|
||||||
|
t.Fatalf("count traffic %q: %v", email, err)
|
||||||
|
}
|
||||||
|
if n != 0 {
|
||||||
|
t.Fatalf("traffic rows for %q = %d, want 0", email, n)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestNodeSyncInboundRemovalStaysOnOneConnection(t *testing.T) {
|
||||||
|
db := initTrafficTestDB(t)
|
||||||
|
goneSettings := `{"clients": [{"email": "gone", "enable": true}]}`
|
||||||
|
keepSettings := `{"clients": [{"email": "keep", "enable": true}]}`
|
||||||
|
createNodeInboundWithSettings(t, db, 1, "n1-gone", 41001, goneSettings)
|
||||||
|
createNodeInboundWithSettings(t, db, 1, "n1-keep", 41002, keepSettings)
|
||||||
|
svc := &InboundService{}
|
||||||
|
keep := &model.Inbound{Tag: "n1-keep", Enable: true, Settings: keepSettings, ClientStats: []xray.ClientTraffic{{Email: "keep", Up: 10, Down: 10, Enable: true}}}
|
||||||
|
if err := syncNodeSnapshot(t, svc, 1,
|
||||||
|
&model.Inbound{Tag: "n1-gone", Enable: true, Settings: goneSettings, ClientStats: []xray.ClientTraffic{{Email: "gone", Up: 10, Down: 10, Enable: true}}},
|
||||||
|
keep); err != nil {
|
||||||
|
t.Fatalf("seed sync: %v", err)
|
||||||
|
}
|
||||||
|
readTraffic(t, db, "gone")
|
||||||
|
|
||||||
|
capPoolAtOneConn(t)
|
||||||
|
err := runWithoutSecondConn(t, "node sync inbound-removal sweep", func() error {
|
||||||
|
return syncNodeSnapshot(t, svc, 1, keep)
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("sync node 1 without n1-gone: %v", err)
|
||||||
|
}
|
||||||
|
assertNoTraffic(t, "gone")
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestNodeSyncClientGoneStaysOnOneConnection(t *testing.T) {
|
||||||
|
db := initTrafficTestDB(t)
|
||||||
|
bothSettings := `{"clients": [{"email": "gone", "enable": true}, {"email": "keep", "enable": true}]}`
|
||||||
|
createNodeInboundWithSettings(t, db, 1, "n1-in", 41001, bothSettings)
|
||||||
|
svc := &InboundService{}
|
||||||
|
if err := syncNodeSnapshot(t, svc, 1, &model.Inbound{Tag: "n1-in", Enable: true, Settings: bothSettings, ClientStats: []xray.ClientTraffic{
|
||||||
|
{Email: "gone", Up: 10, Down: 10, Enable: true},
|
||||||
|
{Email: "keep", Up: 10, Down: 10, Enable: true},
|
||||||
|
}}); err != nil {
|
||||||
|
t.Fatalf("seed sync: %v", err)
|
||||||
|
}
|
||||||
|
readTraffic(t, db, "gone")
|
||||||
|
|
||||||
|
capPoolAtOneConn(t)
|
||||||
|
err := runWithoutSecondConn(t, "node sync per-client sweep", func() error {
|
||||||
|
return syncNodeSnapshot(t, svc, 1, &model.Inbound{
|
||||||
|
Tag: "n1-in",
|
||||||
|
Enable: true,
|
||||||
|
Settings: `{"clients": [{"email": "keep", "enable": true}]}`,
|
||||||
|
ClientStats: []xray.ClientTraffic{{Email: "keep", Up: 20, Down: 20, Enable: true}},
|
||||||
|
})
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("sync node 1 without the gone client: %v", err)
|
||||||
|
}
|
||||||
|
var baselines int64
|
||||||
|
if err := db.Model(&model.NodeClientTraffic{}).Where("node_id = ? AND email = ?", 1, "gone").Count(&baselines).Error; err != nil {
|
||||||
|
t.Fatalf("count node baselines: %v", err)
|
||||||
|
}
|
||||||
|
if baselines != 0 {
|
||||||
|
t.Fatalf("node baseline for the gone client = %d rows, want 0", baselines)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestUpdateClientTrafficsStaysOnOneConnection(t *testing.T) {
|
||||||
|
setupBulkDB(t)
|
||||||
|
svc := &InboundService{}
|
||||||
|
clients := []model.Client{{Email: "gone@x", ID: "aaaaaaaa-0000-0000-0000-000000000101", Enable: true}}
|
||||||
|
oldIb := mkInbound(t, 22101, model.VLESS, clientsSettings(t, clients))
|
||||||
|
if err := database.GetDB().Create(&xray.ClientTraffic{InboundId: oldIb.Id, Email: "gone@x", Enable: true}).Error; err != nil {
|
||||||
|
t.Fatalf("seed traffic: %v", err)
|
||||||
|
}
|
||||||
|
newIb := *oldIb
|
||||||
|
newIb.Settings = `{"clients": []}`
|
||||||
|
|
||||||
|
capPoolAtOneConn(t)
|
||||||
|
err := runWithoutSecondConn(t, "inbound edit dropping a client", func() error {
|
||||||
|
return database.GetDB().Transaction(func(tx *gorm.DB) error {
|
||||||
|
return svc.updateClientTraffics(tx, oldIb, &newIb)
|
||||||
|
})
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("updateClientTraffics: %v", err)
|
||||||
|
}
|
||||||
|
var n int64
|
||||||
|
database.GetDB().Model(xray.ClientTraffic{}).Where("email = ?", "gone@x").Count(&n)
|
||||||
|
if n != 0 {
|
||||||
|
t.Fatalf("traffic rows for dropped client = %d, want 0", n)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestUpdateInboundClientRenameOntoExistingStatStaysOnOneConnection(t *testing.T) {
|
||||||
|
setupBulkDB(t)
|
||||||
|
svc := &ClientService{}
|
||||||
|
inboundSvc := &InboundService{}
|
||||||
|
source := []model.Client{{Email: "old@x", ID: "aaaaaaaa-0000-0000-0000-000000000102", SubID: "sub-102", Enable: true}}
|
||||||
|
ib := mkInbound(t, 22102, model.VLESS, clientsSettings(t, source))
|
||||||
|
if err := svc.SyncInbound(nil, ib.Id, source); err != nil {
|
||||||
|
t.Fatalf("seed linkage: %v", err)
|
||||||
|
}
|
||||||
|
if err := database.GetDB().Create(&xray.ClientTraffic{InboundId: ib.Id, Email: "new@x", Enable: true}).Error; err != nil {
|
||||||
|
t.Fatalf("seed target traffic: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
renamed := source
|
||||||
|
renamed[0].Email = "new@x"
|
||||||
|
capPoolAtOneConn(t)
|
||||||
|
err := runWithoutSecondConn(t, "client rename onto an existing stats row", func() error {
|
||||||
|
_, err := svc.UpdateInboundClient(inboundSvc, &model.Inbound{Id: ib.Id, Settings: clientsSettings(t, renamed)}, "old@x")
|
||||||
|
return err
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("UpdateInboundClient: %v", err)
|
||||||
|
}
|
||||||
|
var n int64
|
||||||
|
database.GetDB().Model(xray.ClientTraffic{}).Where("email = ?", "old@x").Count(&n)
|
||||||
|
if n != 0 {
|
||||||
|
t.Fatalf("traffic rows for renamed-away email = %d, want 0", n)
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user