fix(clients): push bulk client changes to nodes only after the commit lands (#6181)

* fix(clients): apply bulk mutations after durable commit * test(clients): guard bulk pushes behind commit * fix(clients): fully delete remote bulk clients --------- Co-authored-by: n0ctal <293235942+n0ctal@users.noreply.github.com>
2026-08-15 07:40:59 +00:00 · 2026-08-14 23:11:19 +05:00
parent b56b087254
commit c5dec64d36
2 changed files with 137 additions and 91 deletions
@@ -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
@@ -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)