diff --git a/internal/web/service/client_bulk.go b/internal/web/service/client_bulk.go index f48fd02e9..45030d466 100644 --- a/internal/web/service/client_bulk.go +++ b/internal/web/service/client_bulk.go @@ -1143,7 +1143,7 @@ func (s *ClientService) bulkDelInboundClients( var sharedSet map[string]bool if !keepTraffic { var sharedErr error - sharedSet, sharedErr = inboundSvc.emailsUsedByOtherInbounds(foundList, inboundId) + sharedSet, sharedErr = inboundSvc.emailsUsedByOtherInbounds(db, foundList, inboundId) if sharedErr != nil { for email := range wantedEmails { res.perEmailSkipped[email] = sharedErr.Error() diff --git a/internal/web/service/client_identity_normalized_test.go b/internal/web/service/client_identity_normalized_test.go index ff5fd47fc..5b1906715 100644 --- a/internal/web/service/client_identity_normalized_test.go +++ b/internal/web/service/client_identity_normalized_test.go @@ -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 { t.Fatalf("emailsUsedByOtherInbounds: %v", err) } if !shared["alice@x"] { 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 { t.Fatalf("emailUsedByOtherInbounds: %v", err) } diff --git a/internal/web/service/client_inbound_apply.go b/internal/web/service/client_inbound_apply.go index 3003759db..84a3f9849 100644 --- a/internal/web/service/client_inbound_apply.go +++ b/internal/web/service/client_inbound_apply.go @@ -136,7 +136,7 @@ func (s *ClientService) delInboundClients(inboundSvc *InboundService, inboundId } } var sharedErr error - sharedSet, sharedErr = inboundSvc.emailsUsedByOtherInbounds(removedEmails, inboundId) + sharedSet, sharedErr = inboundSvc.emailsUsedByOtherInbounds(db, removedEmails, inboundId) if sharedErr != nil { return false, sharedErr } @@ -946,7 +946,7 @@ func (s *ClientService) UpdateInboundClient(inboundSvc *InboundService, data *mo return e } } else { - stillUsed, sErr := inboundSvc.emailUsedByOtherInbounds(oldEmail, data.Id) + stillUsed, sErr := inboundSvc.emailUsedByOtherInbounds(tx, oldEmail, data.Id) if sErr != nil { return sErr } @@ -968,7 +968,7 @@ func (s *ClientService) UpdateInboundClient(inboundSvc *InboundService, data *mo } } } else { - stillUsed, sErr := inboundSvc.emailUsedByOtherInbounds(oldEmail, data.Id) + stillUsed, sErr := inboundSvc.emailUsedByOtherInbounds(tx, oldEmail, data.Id) if sErr != nil { return sErr } @@ -1140,7 +1140,7 @@ func (s *ClientService) DelInboundClientByEmail(inboundSvc *InboundService, inbo prevSettings := oldInbound.Settings oldInbound.Settings = string(newSettings) - emailShared, err := inboundSvc.emailUsedByOtherInbounds(email, inboundId) + emailShared, err := inboundSvc.emailUsedByOtherInbounds(db, email, inboundId) if err != nil { return false, err } diff --git a/internal/web/service/inbound.go b/internal/web/service/inbound.go index 7c763018f..8c87ed1ca 100644 --- a/internal/web/service/inbound.go +++ b/internal/web/service/inbound.go @@ -2290,7 +2290,7 @@ func (s *InboundService) updateClientTraffics(tx *gorm.DB, oldInbound *model.Inb if _, kept := newEmails[email]; kept { continue } - stillUsed, err := s.emailUsedByOtherInbounds(email, oldInbound.Id) + stillUsed, err := s.emailUsedByOtherInbounds(tx, email, oldInbound.Id) if err != nil { return err } diff --git a/internal/web/service/inbound_clients.go b/internal/web/service/inbound_clients.go index 0391affdb..96a945cc0 100644 --- a/internal/web/service/inbound_clients.go +++ b/internal/web/service/inbound_clients.go @@ -124,14 +124,14 @@ func (s *InboundService) backfillClientStats(db *gorm.DB, inbounds []*model.Inbo return clientsByInbound } -// emailUsedByOtherInbounds reports whether email lives in any inbound other -// than exceptInboundId. Empty email returns false. -func (s *InboundService) emailUsedByOtherInbounds(email string, exceptInboundId int) (bool, error) { +// emailUsedByOtherInbounds reports whether email lives in any inbound other than +// exceptInboundId. Inside a transaction pass tx: a pool query there starves the pool. +func (s *InboundService) emailUsedByOtherInbounds(db *gorm.DB, email string, exceptInboundId int) (bool, error) { if email == "" { return false, nil } var count int64 - err := database.GetDB().Table("client_inbounds"). + err := db.Table("client_inbounds"). Joins("JOIN clients ON clients.id = client_inbounds.client_id"). Where("client_inbounds.inbound_id != ? AND LOWER(clients.email) = ?", exceptInboundId, strings.ToLower(strings.TrimSpace(email))). @@ -142,7 +142,7 @@ func (s *InboundService) emailUsedByOtherInbounds(email string, exceptInboundId 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)) want := make(map[string]struct{}, len(emails)) for _, e := range emails { @@ -158,7 +158,6 @@ func (s *InboundService) emailsUsedByOtherInbounds(emails []string, exceptInboun for e := range want { lowered = append(lowered, e) } - db := database.GetDB() for _, batch := range chunkStrings(lowered, sqlInChunk) { var rows []struct{ Email string } err := db.Table("client_inbounds"). diff --git a/internal/web/service/inbound_node.go b/internal/web/service/inbound_node.go index 327659ef1..f481d2c05 100644 --- a/internal/web/service/inbound_node.go +++ b/internal/web/service/inbound_node.go @@ -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 // loses the summed history, and the next node sync would re-seed the // 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 { 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: // 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 { return false, uErr } diff --git a/internal/web/service/tx_pool_starvation_test.go b/internal/web/service/tx_pool_starvation_test.go new file mode 100644 index 000000000..e14455a6f --- /dev/null +++ b/internal/web/service/tx_pool_starvation_test.go @@ -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) + } +}