diff --git a/internal/web/service/client_bulk.go b/internal/web/service/client_bulk.go index 23d62c3db..9a2831f4e 100644 --- a/internal/web/service/client_bulk.go +++ b/internal/web/service/client_bulk.go @@ -652,7 +652,6 @@ func (s *ClientService) bulkAdjustInboundClients( } return res } - prevSettings := oldInbound.Settings oldInbound.Settings = string(newSettings) // A flow change rewrites the user's xray config, which the lightweight @@ -662,45 +661,6 @@ func (s *ClientService) bulkAdjustInboundClients( res.needRestart = true } - if oldInbound.NodeID != nil { - rt, push, _, perr := inboundSvc.nodePushPlan(oldInbound) - if perr != nil { - for email := range foundEmails { - res.perEmailSkipped[email] = perr.Error() - delete(foundEmails, email) - } - } else { - if flowChanged { - push = false - } - // Large batches collapse into one reconcile push rather than M updates. - if push && len(foundEmails) > nodeBulkPushThreshold { - push = false - } - if push { - pushFailed := false - for email := range foundEmails { - entry := plan[email] - updated := *entry.record.ToClient() - if entry.applyExpiry { - updated.ExpiryTime = entry.newExpiry - } - if entry.applyTotal { - updated.TotalGB = entry.newTotal - } - updated.UpdatedAt = nowMs - if err1 := rt.UpdateUser(context.Background(), oldInbound, email, updated); err1 != nil { - logger.Warning("Error in updating client on", rt.Name(), ":", err1) - pushFailed = true - } - } - if !pushFailed { - advancePushedInbound(rt, prevSettings, oldInbound) - } - } - } - } - // Serialize against the traffic poll to avoid the cross-transaction // lock-order deadlock on inbounds/client_records (runSerializedTx). txErr := runSerializedTx(func(tx *gorm.DB) error { @@ -725,6 +685,26 @@ func (s *ClientService) bulkAdjustInboundClients( res.perEmailSkipped[email] = txErr.Error() } } + } else if oldInbound.NodeID != nil && !flowChanged && len(foundEmails) <= nodeBulkPushThreshold { + rt, push, _, perr := inboundSvc.nodePushPlan(oldInbound) + if perr != nil { + logger.Warning("BulkAdjust: node runtime lookup after commit failed:", perr) + } else if push { + for email := range foundEmails { + entry := plan[email] + updated := *entry.record.ToClient() + if entry.applyExpiry { + updated.ExpiryTime = entry.newExpiry + } + if entry.applyTotal { + updated.TotalGB = entry.newTotal + } + updated.UpdatedAt = nowMs + if err1 := rt.UpdateUser(context.Background(), oldInbound, email, updated); err1 != nil { + logger.Warning("Error in updating client on", rt.Name(), ":", err1) + } + } + } } return res @@ -980,7 +960,6 @@ func (s *ClientService) bulkDelInboundClients( } return res } - prevSettings := oldInbound.Settings oldInbound.Settings = string(newSettings) foundList := make([]string, 0, len(foundEmails)) @@ -1048,56 +1027,6 @@ func (s *ClientService) bulkDelInboundClients( } } - if oldInbound.NodeID == nil { - rt, rterr := inboundSvc.runtimeFor(oldInbound) - if rterr != nil { - res.needRestart = true - } else { - for email := range foundEmails { - if !enableByEmail[email] || !notDepletedByEmail[email] { - continue - } - err1 := rt.RemoveUser(context.Background(), oldInbound, email) - if err1 == nil { - logger.Debug("Client deleted on", rt.Name(), ":", email) - } else if strings.Contains(err1.Error(), fmt.Sprintf("User %s not found.", email)) { - logger.Debug("User is already deleted. Nothing to do more...") - } else { - logger.Debug("Error in deleting client on", rt.Name(), ":", err1) - res.needRestart = true - } - } - } - } else { - rt, push, _, perr := inboundSvc.nodePushPlan(oldInbound) - if perr != nil { - for email := range foundEmails { - res.perEmailSkipped[email] = perr.Error() - delete(foundEmails, email) - } - } else { - // Large batches collapse into one reconcile push rather than M deletes. - if push && len(foundEmails) > nodeBulkPushThreshold { - push = false - } - if push { - // bulkDelInboundClients only runs for full client deletion - // (BulkDelete), so the node must drop its client record too, - // not just detach from this inbound (#5797). - pushFailed := false - for email := range foundEmails { - if err1 := rt.DeleteClient(context.Background(), email); err1 != nil { - logger.Warning("Error in deleting client on", rt.Name(), ":", err1) - pushFailed = true - } - } - if !pushFailed { - advancePushedInbound(rt, prevSettings, oldInbound) - } - } - } - } - // Serialize against the traffic poll to avoid the cross-transaction // lock-order deadlock on inbounds/client_records (runSerializedTx). txErr := runSerializedTx(func(tx *gorm.DB) error { @@ -1122,6 +1051,37 @@ func (s *ClientService) bulkDelInboundClients( res.perEmailSkipped[email] = txErr.Error() } } + } else if oldInbound.NodeID == nil { + rt, rterr := inboundSvc.runtimeFor(oldInbound) + if rterr != nil { + res.needRestart = true + } else { + for email := range foundEmails { + if !enableByEmail[email] || !notDepletedByEmail[email] { + continue + } + err1 := rt.RemoveUser(context.Background(), oldInbound, email) + if err1 == nil { + logger.Debug("Client deleted on", rt.Name(), ":", email) + } else if strings.Contains(err1.Error(), fmt.Sprintf("User %s not found.", email)) { + logger.Debug("User is already deleted. Nothing to do more...") + } else { + logger.Debug("Error in deleting client on", rt.Name(), ":", err1) + res.needRestart = true + } + } + } + } else if len(foundEmails) <= nodeBulkPushThreshold { + rt, push, _, perr := inboundSvc.nodePushPlan(oldInbound) + if perr != nil { + logger.Warning("BulkDelete: node runtime lookup after commit failed:", perr) + } else if push { + for email := range foundEmails { + if err1 := rt.DeleteClient(context.Background(), email); err1 != nil { + logger.Warning("Error in deleting client on", rt.Name(), ":", err1) + } + } + } } return res diff --git a/internal/web/service/node_bulk_dispatch_test.go b/internal/web/service/node_bulk_dispatch_test.go index 3e071e4b4..7c26bdfbb 100644 --- a/internal/web/service/node_bulk_dispatch_test.go +++ b/internal/web/service/node_bulk_dispatch_test.go @@ -2,11 +2,13 @@ package service import ( "context" + "errors" "fmt" "sync/atomic" "testing" "github.com/google/uuid" + "gorm.io/gorm" "github.com/mhsanaei/3x-ui/v3/internal/database" "github.com/mhsanaei/3x-ui/v3/internal/database/model" @@ -165,6 +167,90 @@ func TestNodeBulk_SmallAddPushesLive(t *testing.T) { } } +func TestNodeBulkAdjustDoesNotPushBeforeFailedCommit(t *testing.T) { + setupBulkDB(t) + nodeID, fake := setupNodeRuntime(t) + client := model.Client{ + ID: uuid.NewString(), + Email: "txfail-adjust@x", + Enable: true, + ExpiryTime: 1_900_000_000_000, + } + nodeInbound(t, nodeID, 30022, []model.Client{client}) + + db := database.GetDB() + const callbackName = "bulk-adjust:fail-inbound-update" + if err := db.Callback().Update().After("gorm:update").Register(callbackName, func(tx *gorm.DB) { + if tx.Statement != nil && tx.Statement.Table == "inbounds" { + tx.AddError(errors.New("injected bulk-adjust transaction failure")) + } + }); err != nil { + t.Fatalf("register callback: %v", err) + } + t.Cleanup(func() { _ = db.Callback().Update().Remove(callbackName) }) + + result, _, err := (&ClientService{}).BulkAdjust(&InboundService{}, []string{client.Email}, 1, 0, "") + if err != nil { + t.Fatalf("BulkAdjust: %v", err) + } + if result.Adjusted != 0 || len(result.Skipped) != 1 { + t.Fatalf("BulkAdjust result = %+v, want one skipped client after injected failure", result) + } + if got := fake.updateUser.Load(); got != 0 { + t.Fatalf("failed transaction pushed %d UpdateUser call(s) to the node, want 0", got) + } +} + +func TestNodeBulkDeleteDoesNotPushBeforeFailedCommit(t *testing.T) { + setupBulkDB(t) + nodeID, fake := setupNodeRuntime(t) + client := model.Client{ID: uuid.NewString(), Email: "txfail-delete@x", Enable: true} + nodeInbound(t, nodeID, 30023, []model.Client{client}) + + db := database.GetDB() + const callbackName = "bulk-delete:fail-inbound-update" + if err := db.Callback().Update().After("gorm:update").Register(callbackName, func(tx *gorm.DB) { + if tx.Statement != nil && tx.Statement.Table == "inbounds" { + tx.AddError(errors.New("injected bulk-delete transaction failure")) + } + }); err != nil { + t.Fatalf("register callback: %v", err) + } + t.Cleanup(func() { _ = db.Callback().Update().Remove(callbackName) }) + + result, _, err := (&ClientService{}).BulkDelete(&InboundService{}, []string{client.Email}, true) + if err != nil { + t.Fatalf("BulkDelete: %v", err) + } + if result.Deleted != 0 || len(result.Skipped) != 1 { + t.Fatalf("BulkDelete result = %+v, want one skipped client after injected failure", result) + } + if got := fake.deleteClient.Load() + fake.deleteUser.Load(); got != 0 { + t.Fatalf("failed transaction pushed %d delete call(s) to the node, want 0", got) + } +} + +func TestNodeBulkSmallDeleteRemovesWholeRemoteClient(t *testing.T) { + setupBulkDB(t) + nodeID, fake := setupNodeRuntime(t) + client := model.Client{ID: uuid.NewString(), Email: "full-delete@x", Enable: true} + nodeInbound(t, nodeID, 30024, []model.Client{client}) + + result, _, err := (&ClientService{}).BulkDelete(&InboundService{}, []string{client.Email}, true) + if err != nil { + t.Fatalf("BulkDelete: %v", err) + } + if result.Deleted != 1 || len(result.Skipped) != 0 { + t.Fatalf("BulkDelete result = %+v, want one deleted client", result) + } + if got := fake.deleteClient.Load(); got != 1 { + t.Fatalf("remote DeleteClient calls = %d, want 1", got) + } + if got := fake.deleteUser.Load(); got != 0 { + t.Fatalf("remote DeleteUser detach calls = %d, want 0 for full deletion", got) + } +} + func TestNodeUpdateInboundClientNoopSkipsRuntimeAndDirty(t *testing.T) { setupBulkDB(t) nodeID, fake := setupNodeRuntime(t)