fix(traffic): apply maintenance side effects only after the commit lands (#6200)

* fix(traffic): apply maintenance after durable commit

* fix(traffic): apply all runtime maintenance after commit

---------

Co-authored-by: n0ctal <293235942+n0ctal@users.noreply.github.com>

frontend/src/generated/types.ts

@@ -4,6 +4,7 @@ export type ProcessState = string;
 export type Protocol = string;
 export type SubLinkProvider = unknown;
 export type staticEgressResolver = string;
+export type trafficLocalApplyAction = number;
 export type transportBits = number;
 
 export interface AllSetting {

frontend/src/generated/zod.ts

@@ -15,6 +15,9 @@ export type SubLinkProvider = z.infer<typeof SubLinkProviderSchema>;
 export const staticEgressResolverSchema = z.string();
 export type staticEgressResolver = z.infer<typeof staticEgressResolverSchema>;
 
+export const trafficLocalApplyActionSchema = z.number().int();
+export type trafficLocalApplyAction = z.infer<typeof trafficLocalApplyActionSchema>;
+
 export const transportBitsSchema = z.number().int();
 export type transportBits = z.infer<typeof transportBitsSchema>;
 

internal/web/service/global_traffic_test.go

@@ -75,7 +75,7 @@ func TestDepletedCond_ProbeGuard(t *testing.T) {
 		t.Fatalf("empty globals must use the local-only predicate")
 	}
 	seedClientRow(t, "local-cap", 1, 600, 600, 1000)
-	if _, count, err := svc.disableInvalidClients(db); err != nil {
+	if _, count, _, err := svc.disableInvalidClients(db, newTrafficMutationBatch()); err != nil {
 		t.Fatalf("disableInvalidClients: %v", err)
 	} else if count != 1 {
 		t.Fatalf("local over-quota client must be disabled, disabled %d", count)
@@ -115,7 +115,7 @@ func TestStaleGlobalTraffic_Ignored(t *testing.T) {
 		if got, _ := depletedCond(db); got != depletedClientsCondLocal {
 			t.Fatalf("only stale globals must fall back to the local-only predicate")
 		}
-		if _, count, err := svc.disableInvalidClients(db); err != nil {
+		if _, count, _, err := svc.disableInvalidClients(db, newTrafficMutationBatch()); err != nil {
 			t.Fatalf("disableInvalidClients: %v", err)
 		} else if count != 0 {
 			t.Fatalf("stale global usage must not disable a client, disabled %d", count)
@@ -140,7 +140,7 @@ func TestStaleGlobalTraffic_Ignored(t *testing.T) {
 		if got, _ := depletedCond(db); got != depletedClientsCond {
 			t.Fatalf("a fresh global row must select the cross-panel predicate")
 		}
-		if _, count, err := svc.disableInvalidClients(db); err != nil {
+		if _, count, _, err := svc.disableInvalidClients(db, newTrafficMutationBatch()); err != nil {
 			t.Fatalf("disableInvalidClients: %v", err)
 		} else if count != 0 {
 			t.Fatalf("the live master reports usage well under quota, disabled %d", count)
@@ -149,7 +149,7 @@ func TestStaleGlobalTraffic_Ignored(t *testing.T) {
 		if err := svc.AcceptGlobalTraffic("live-master", []*xray.ClientTraffic{{Email: "cap", Up: 600, Down: 500}}); err != nil {
 			t.Fatalf("AcceptGlobalTraffic: %v", err)
 		}
-		if _, count, err := svc.disableInvalidClients(db); err != nil {
+		if _, count, _, err := svc.disableInvalidClients(db, newTrafficMutationBatch()); err != nil {
 			t.Fatalf("disableInvalidClients: %v", err)
 		} else if count != 1 {
 			t.Fatalf("fresh cross-panel depletion must disable the client, disabled %d", count)
@@ -167,7 +167,7 @@ func TestGlobalUsage_DisablesClient(t *testing.T) {
 		t.Fatalf("AcceptGlobalTraffic: %v", err)
 	}
 
-	if _, count, err := svc.disableInvalidClients(db); err != nil {
+	if _, count, _, err := svc.disableInvalidClients(db, newTrafficMutationBatch()); err != nil {
 		t.Fatalf("disableInvalidClients: %v", err)
 	} else if count != 1 {
 		t.Fatalf("expected 1 client disabled, got %d", count)

internal/web/service/inbound.go

@@ -29,7 +29,6 @@ import (
 )
 
 type InboundService struct {
-	xrayApi         xray.XrayAPI
 	clientService   ClientService
 	fallbackService FallbackService
 }

internal/web/service/inbound_autorenew_shadowsocks_test.go

@@ -77,7 +77,7 @@ func TestAutoRenewShadowsocksKeepsSettingsClean(t *testing.T) {
 		t.Fatalf("seed client_traffics: %v", err)
 	}
 
-	if _, count, err := svc.autoRenewClients(db); err != nil {
+	if _, count, err := svc.autoRenewClients(db, newTrafficMutationBatch()); err != nil {
 		t.Fatalf("autoRenewClients: %v", err)
 	} else if count != 1 {
 		t.Fatalf("renewed count = %d, want 1", count)

internal/web/service/inbound_autorenew_test.go

@@ -52,7 +52,7 @@ func TestAutoRenewClients_MultiInbound(t *testing.T) {
 		t.Fatalf("seed client_traffics: %v", err)
 	}
 
-	if _, count, err := svc.autoRenewClients(db); err != nil {
+	if _, count, err := svc.autoRenewClients(db, newTrafficMutationBatch()); err != nil {
 		t.Fatalf("autoRenewClients: %v", err)
 	} else if count != 3 {
 		t.Fatalf("renewed count = %d, want 3", count)

internal/web/service/inbound_disable.go

@@ -1,44 +1,27 @@
 package service
 
 import (
-	"context"
 	"encoding/json"
-	"fmt"
 	"slices"
-	"strings"
 	"time"
 
 	"github.com/mhsanaei/3x-ui/v3/internal/database/model"
-	"github.com/mhsanaei/3x-ui/v3/internal/logger"
 	"github.com/mhsanaei/3x-ui/v3/internal/xray"
 
 	"gorm.io/gorm"
 )
 
-func (s *InboundService) disableInvalidInbounds(tx *gorm.DB) (bool, int64, error) {
+func (s *InboundService) disableInvalidInbounds(tx *gorm.DB, mutationBatch *trafficMutationBatch) (bool, int64, error) {
 	now := time.Now().Unix() * 1000
-	needRestart := false
-
-	if process := currentXrayProcess(); process != nil {
-		var tags []string
-		err := tx.Table("inbounds").
-			Select("inbounds.tag").
-			Where("((total > 0 and up + down >= total) or (expiry_time > 0 and expiry_time <= ?)) and enable = ? and node_id IS NULL", now, true).
-			Scan(&tags).Error
-		if err != nil {
-			return false, 0, err
-		}
-		_ = s.xrayApi.Init(process.GetAPIPort())
-		for _, tag := range tags {
-			err1 := s.xrayApi.DelInbound(tag)
-			if err1 == nil {
-				logger.Debug("Inbound disabled by api:", tag)
-			} else {
-				logger.Debug("Error in disabling inbound by api:", err1)
-				needRestart = true
-			}
-		}
-		s.xrayApi.Close()
+	var inbounds []model.Inbound
+	if err := tx.Where("((total > 0 and up + down >= total) or (expiry_time > 0 and expiry_time <= ?)) and enable = ? and node_id IS NULL", now, true).
+		Find(&inbounds).Error; err != nil {
+		return false, 0, err
+	}
+	for i := range inbounds {
+		mutationBatch.localPlans = append(mutationBatch.localPlans, trafficLocalApplyPlan{
+			action: trafficDisableInbound, inbound: inbounds[i],
+		})
 	}
 
 	result := tx.Model(model.Inbound{}).
@@ -46,7 +29,7 @@ func (s *InboundService) disableInvalidInbounds(tx *gorm.DB) (bool, int64, error
 		Update("enable", false)
 	err := result.Error
 	count := result.RowsAffected
-	return needRestart, count, err
+	return false, count, err
 }
 
 const globalTrafficFreshWindow = 24 * time.Hour
@@ -94,8 +77,8 @@ func depletedCond(tx *gorm.DB) (string, []any) {
 	return depletedClientsCondLocal, []any{now}
 }
 
-func (s *InboundService) disableInvalidClients(tx *gorm.DB) (bool, int64, error) {
-	needRestart := false
+func (s *InboundService) disableInvalidClients(tx *gorm.DB, mutationBatch *trafficMutationBatch) (bool, int64, []int, error) {
+	now := time.Now().UnixMilli()
 	cond, condArgs := depletedCond(tx)
 
 	var depletedRows []xray.ClientTraffic
@@ -103,10 +86,10 @@ func (s *InboundService) disableInvalidClients(tx *gorm.DB) (bool, int64, error)
 		Where(cond+" AND enable = ?", append(condArgs, true)...).
 		Find(&depletedRows).Error
 	if err != nil {
-		return false, 0, err
+		return false, 0, nil, err
 	}
 	if len(depletedRows) == 0 {
-		return false, 0, nil
+		return false, 0, nil, nil
 	}
 
 	depletedEmails := make([]string, 0, len(depletedRows))
@@ -134,47 +117,39 @@ func (s *InboundService) disableInvalidClients(tx *gorm.DB) (bool, int64, error)
 			WHERE clients.email IN ?
 		`, depletedEmails).Scan(&targets).Error
 		if err != nil {
-			return false, 0, err
+			return false, 0, nil, err
 		}
 	}
 
-	var localTargets []target
-	localByInbound := make(map[int]map[string]struct{})
-	remoteByInbound := make(map[int][]target)
+	byInbound := make(map[int][]target)
 	for _, t := range targets {
-		if t.NodeID == nil {
-			localTargets = append(localTargets, t)
-			if localByInbound[t.InboundID] == nil {
-				localByInbound[t.InboundID] = make(map[string]struct{})
-			}
-			localByInbound[t.InboundID][t.Email] = struct{}{}
-		} else {
-			remoteByInbound[t.InboundID] = append(remoteByInbound[t.InboundID], t)
-		}
+		byInbound[t.InboundID] = append(byInbound[t.InboundID], t)
 	}
 
-	if process := currentXrayProcess(); process != nil && len(localTargets) > 0 {
-		_ = s.xrayApi.Init(process.GetAPIPort())
-		for _, t := range localTargets {
-			err1 := s.xrayApi.RemoveUser(t.Tag, t.Email)
-			if err1 == nil {
-				logger.Debug("Client disabled by api:", t.Email)
-			} else if strings.Contains(err1.Error(), fmt.Sprintf("User %s not found.", t.Email)) {
-				logger.Debug("User is already disabled. Nothing to do more...")
-			} else {
-				logger.Debug("Error in disabling client by api:", err1)
-				needRestart = true
-			}
+	disabledNodeIDs := make(map[int]struct{})
+	for inboundID, group := range byInbound {
+		emails := make(map[string]struct{}, len(group))
+		for _, t := range group {
+			emails[t.Email] = struct{}{}
 		}
-		s.xrayApi.Close()
-	}
-
-	for inboundID, emails := range localByInbound {
-		if _, _, mErr := s.markClientsDisabledInSettings(tx, inboundID, emails); mErr != nil {
-			logger.Warning("disableInvalidClients: settings.JSON sync failed for inbound", inboundID, ":", mErr)
+		oldInbound, inbound, mErr := s.markClientsDisabledInSettings(tx, inboundID, emails)
+		if mErr != nil {
+			return false, 0, nil, mErr
+		}
+		if inbound.NodeID != nil {
+			mutationBatch.remotePlans = append(mutationBatch.remotePlans, trafficInboundUpdatePlan{
+				oldInbound: *oldInbound, newInbound: *inbound,
+			})
+			mutationBatch.addNode(*inbound.NodeID)
+			disabledNodeIDs[*inbound.NodeID] = struct{}{}
+			continue
+		}
+		for email := range emails {
+			mutationBatch.localPlans = append(mutationBatch.localPlans, trafficLocalApplyPlan{
+				action: trafficRemoveUser, inbound: *inbound, email: email,
+			})
 		}
 	}
-
 	// Flip the rows already collected above by primary key instead of
 	// re-evaluating the depleted predicate, which was a second full scan of
 	// client_traffics on every poll. Sorted ids keep the lock order stable.
@@ -189,7 +164,7 @@ func (s *InboundService) disableInvalidClients(tx *gorm.DB) (bool, int64, error)
 			Where("id IN ? AND enable = ?", batch, true).
 			Update("enable", false)
 		if result.Error != nil {
-			return needRestart, count, result.Error
+			return false, count, nil, result.Error
 		}
 		count += result.RowsAffected
 	}
@@ -197,23 +172,17 @@ func (s *InboundService) disableInvalidClients(tx *gorm.DB) (bool, int64, error)
 	if len(depletedEmails) > 0 {
 		if err := tx.Model(&model.ClientRecord{}).
 			Where("email IN ?", depletedEmails).
-			Updates(map[string]any{"enable": false, "updated_at": time.Now().UnixMilli()}).Error; err != nil {
-			logger.Warning("disableInvalidClients update clients.enable:", err)
+			Updates(map[string]any{"enable": false, "updated_at": now}).Error; err != nil {
+			return false, count, nil, err
 		}
 	}
 
-	for inboundID, group := range remoteByInbound {
-		emails := make(map[string]struct{}, len(group))
-		for _, t := range group {
-			emails[t.Email] = struct{}{}
-		}
-		if pushErr := s.disableRemoteClients(tx, inboundID, emails); pushErr != nil {
-			logger.Warning("disableInvalidClients: push to remote failed for inbound", inboundID, ":", pushErr)
-			needRestart = true
-		}
+	nodeIDs := make([]int, 0, len(disabledNodeIDs))
+	for nodeID := range disabledNodeIDs {
+		nodeIDs = append(nodeIDs, nodeID)
 	}
 
-	return needRestart, count, nil
+	return false, count, nodeIDs, nil
 }
 
 // markClientsDisabledInSettings flips client.enable=false in the inbound's
@@ -265,23 +234,3 @@ func (s *InboundService) markClientsDisabledInSettings(tx *gorm.DB, inboundID in
 	}
 	return &snapshot, &ib, nil
 }
-
-// disableRemoteClients flips the clients off in the inbound's stored settings
-// and pushes the updated inbound to its node, which applies it to its own
-// running Xray. That push is the whole reconcile — restarting the node's Xray
-// afterwards would drop every live connection on the node for nothing (#5740).
-func (s *InboundService) disableRemoteClients(tx *gorm.DB, inboundID int, emails map[string]struct{}) error {
-	oldSnapshot, ib, err := s.markClientsDisabledInSettings(tx, inboundID, emails)
-	if err != nil {
-		return err
-	}
-
-	rt, err := s.runtimeFor(ib)
-	if err != nil {
-		return err
-	}
-	if err := rt.UpdateInbound(context.Background(), oldSnapshot, ib); err != nil {
-		return err
-	}
-	return nil
-}

internal/web/service/inbound_node.go

@@ -1079,6 +1079,28 @@ func (s *InboundService) setRemoteTrafficLocked(nodeID int, snap *runtime.Traffi
 	return structuralChange, nil
 }
 
+func (s *InboundService) restartRemoteNodesOnDisable(nodeIDs []int) {
+	restartOnDisable, err := (&SettingService{}).GetRestartXrayOnClientDisable()
+	if err != nil {
+		logger.Warning("disableInvalidClients: get RestartXrayOnClientDisable failed:", err)
+		return
+	}
+	if !restartOnDisable {
+		return
+	}
+	for _, nodeID := range nodeIDs {
+		nodeIDCopy := nodeID
+		rt, rtErr := runtime.GetManager().RuntimeFor(&nodeIDCopy)
+		if rtErr != nil {
+			logger.Warning("disableInvalidClients: get runtime for node", nodeID, "failed:", rtErr)
+			continue
+		}
+		if rtErr = rt.RestartXray(context.Background()); rtErr != nil {
+			logger.Warning("disableInvalidClients: restart xray on node", nodeID, "failed:", rtErr)
+		}
+	}
+}
+
 func (s *InboundService) GetOnlineClients() []string {
 	process := currentXrayProcess()
 	if process == nil {

internal/web/service/inbound_traffic.go

@@ -22,60 +22,77 @@ import (
 )
 
 func (s *InboundService) AddTraffic(inboundTraffics []*xray.Traffic, clientTraffics []*xray.ClientTraffic) (needRestart bool, clientsDisabled bool, err error) {
+	var disabledNodeIDs []int
 	err = submitTrafficWrite(func() error {
 		var inner error
-		needRestart, clientsDisabled, inner = s.addTrafficLocked(inboundTraffics, clientTraffics)
+		needRestart, clientsDisabled, disabledNodeIDs, inner = s.addTrafficLocked(inboundTraffics, clientTraffics)
 		return inner
 	})
+	if err == nil && len(disabledNodeIDs) > 0 {
+		s.restartRemoteNodesOnDisable(disabledNodeIDs)
+	}
 	return
 }
 
-func (s *InboundService) addTrafficLocked(inboundTraffics []*xray.Traffic, clientTraffics []*xray.ClientTraffic) (bool, bool, error) {
-	var err error
+func (s *InboundService) addTrafficLocked(inboundTraffics []*xray.Traffic, clientTraffics []*xray.ClientTraffic) (bool, bool, []int, error) {
 	db := database.GetDB()
-	tx := db.Begin()
-
-	defer func() {
-		if err != nil {
-			if rbErr := tx.Rollback().Error; rbErr != nil {
-				logger.Warning("Error rolling back traffic tx:", rbErr)
-			}
-		} else if cErr := tx.Commit().Error; cErr != nil {
-			logger.Warning("Error committing traffic tx:", cErr)
+	// Commit durable traffic before best-effort lifecycle maintenance so helper
+	// failures cannot discard usage already reported by Xray.
+	if err := db.Transaction(func(tx *gorm.DB) error {
+		if err := s.addInboundTraffic(tx, inboundTraffics); err != nil {
+			return err
 		}
-	}()
-	err = s.addInboundTraffic(tx, inboundTraffics)
+		return s.addClientTraffic(tx, clientTraffics)
+	}); err != nil {
+		return false, false, nil, err
+	}
+
+	var (
+		needRestart          bool
+		clientsDisabled      bool
+		disabledNodeIDs      []int
+		disabledClientsCount int64
+	)
+	batch := newTrafficMutationBatch()
+	err := db.Transaction(func(tx *gorm.DB) error {
+		needRestart0, count, err := s.autoRenewClients(tx, batch)
+		if err != nil {
+			return fmt.Errorf("renew clients: %w", err)
+		}
+		if count > 0 {
+			logger.Debugf("%v clients renewed", count)
+		}
+
+		needRestart1, count, nodeIDs, err := s.disableInvalidClients(tx, batch)
+		if err != nil {
+			return fmt.Errorf("disable invalid clients: %w", err)
+		}
+		if count > 0 {
+			logger.Debugf("%v clients disabled", count)
+			disabledClientsCount = count
+		}
+
+		needRestart2, count, err := s.disableInvalidInbounds(tx, batch)
+		if err != nil {
+			return fmt.Errorf("disable invalid inbounds: %w", err)
+		}
+		if count > 0 {
+			logger.Debugf("%v inbounds disabled", count)
+		}
+		if err := batch.markNodesTx(tx); err != nil {
+			return err
+		}
+		needRestart = needRestart0 || needRestart1 || needRestart2
+		clientsDisabled = disabledClientsCount > 0
+		disabledNodeIDs = nodeIDs
+		return nil
+	})
 	if err != nil {
-		return false, false, err
+		logger.Warning("traffic lifecycle maintenance failed after traffic commit:", err)
+		return false, false, nil, nil
 	}
-	err = s.addClientTraffic(tx, clientTraffics)
-	if err != nil {
-		return false, false, err
-	}
-
-	needRestart0, count, renewErr := s.autoRenewClients(tx)
-	if renewErr != nil {
-		logger.Warning("Error in renew clients:", renewErr)
-	} else if count > 0 {
-		logger.Debugf("%v clients renewed", count)
-	}
-
-	disabledClientsCount := int64(0)
-	needRestart1, count, disableClientsErr := s.disableInvalidClients(tx)
-	if disableClientsErr != nil {
-		logger.Warning("Error in disabling invalid clients:", disableClientsErr)
-	} else if count > 0 {
-		logger.Debugf("%v clients disabled", count)
-		disabledClientsCount = count
-	}
-
-	needRestart2, count, disableInboundsErr := s.disableInvalidInbounds(tx)
-	if disableInboundsErr != nil {
-		logger.Warning("Error in disabling invalid inbounds:", disableInboundsErr)
-	} else if count > 0 {
-		logger.Debugf("%v inbounds disabled", count)
-	}
-	return needRestart0 || needRestart1 || needRestart2, disabledClientsCount > 0, nil
+	needRestart = needRestart || s.applyTrafficMutationBatch(batch)
+	return needRestart, clientsDisabled, disabledNodeIDs, nil
 }
 
 func (s *InboundService) addInboundTraffic(tx *gorm.DB, traffics []*xray.Traffic) error {
@@ -304,11 +321,11 @@ func apiUserFromClient(client map[string]any, cipher string) map[string]any {
 	return user
 }
 
-func (s *InboundService) autoRenewClients(tx *gorm.DB) (bool, int64, error) {
+func (s *InboundService) autoRenewClients(tx *gorm.DB, mutationBatch *trafficMutationBatch) (bool, int64, error) {
 	// check for time expired
 	var traffics []*xray.ClientTraffic
 	now := time.Now().Unix() * 1000
-	var err, err1 error
+	var err error
 
 	// Filter to clients that have at least one local inbound. Using
 	// client_traffics.inbound_id is wrong: it goes stale after an inbound is
@@ -335,9 +352,8 @@ func (s *InboundService) autoRenewClients(tx *gorm.DB) (bool, int64, error) {
 	var inbounds []*model.Inbound
 	needRestart := false
 	var clientsToAdd []struct {
-		protocol string
-		tag      string
-		client   map[string]any
+		inbound model.Inbound
+		client  map[string]any
 	}
 
 	// Resolve the inbounds to renew through the client_inbounds link rather than
@@ -408,13 +424,11 @@ func (s *InboundService) autoRenewClients(tx *gorm.DB) (bool, int64, error) {
 				c["enable"] = true
 				clientsToAdd = append(clientsToAdd,
 					struct {
-						protocol string
-						tag      string
-						client   map[string]any
+						inbound model.Inbound
+						client  map[string]any
 					}{
-						protocol: string(inbounds[inbound_index].Protocol),
-						tag:      inbounds[inbound_index].Tag,
-						client:   apiUserFromClient(c, cipher),
+						inbound: *inbounds[inbound_index],
+						client:  apiUserFromClient(c, cipher),
 					})
 			}
 			clients[client_index] = any(c)
@@ -452,18 +466,14 @@ func (s *InboundService) autoRenewClients(tx *gorm.DB) (bool, int64, error) {
 	if err = clearGlobalTraffic(tx, renewEmails...); err != nil {
 		return false, 0, err
 	}
-	if process := currentXrayProcess(); process != nil {
-		err1 = s.xrayApi.Init(process.GetAPIPort())
-		if err1 != nil {
-			return true, int64(len(traffics)), nil
+	for _, clientToAdd := range clientsToAdd {
+		if clientToAdd.inbound.NodeID != nil {
+			mutationBatch.addNode(*clientToAdd.inbound.NodeID)
+			continue
 		}
-		for _, clientToAdd := range clientsToAdd {
-			err1 = s.xrayApi.AddUser(clientToAdd.protocol, clientToAdd.tag, clientToAdd.client)
-			if err1 != nil {
-				needRestart = true
-			}
-		}
-		s.xrayApi.Close()
+		mutationBatch.localPlans = append(mutationBatch.localPlans, trafficLocalApplyPlan{
+			action: trafficAddUser, inbound: clientToAdd.inbound, client: clientToAdd.client,
+		})
 	}
 	return needRestart, int64(len(traffics)), nil
 }
@@ -577,56 +587,58 @@ func (s *InboundService) ResetClientTrafficByEmail(clientEmail string) error {
 }
 
 func (s *InboundService) ResetClientTraffic(id int, clientEmail string) (needRestart bool, err error) {
+	var resetInbound *model.Inbound
 	err = submitTrafficWrite(func() error {
 		var inner error
-		needRestart, inner = s.resetClientTrafficLocked(id, clientEmail)
+		needRestart, resetInbound, inner = s.resetClientTrafficLocked(id, clientEmail)
 		return inner
 	})
 	if err == nil {
 		s.resetMtprotoClientQuota(clientEmail)
+		if resetInbound != nil && resetInbound.NodeID != nil {
+			if rt, rterr := s.runtimeFor(resetInbound); rterr == nil {
+				if e := rt.ResetClientTraffic(context.Background(), resetInbound, clientEmail); e != nil {
+					logger.Warning("ResetClientTraffic: remote propagation to", rt.Name(), "failed:", e)
+				}
+			} else {
+				logger.Warning("ResetClientTraffic: runtime lookup failed:", rterr)
+			}
+		}
 	}
 	return
 }
 
-func (s *InboundService) resetClientTrafficLocked(id int, clientEmail string) (bool, error) {
+func (s *InboundService) resetClientTrafficLocked(id int, clientEmail string) (bool, *model.Inbound, error) {
 	needRestart := false
+	var reenablePlan *trafficLocalApplyPlan
+	var reenableNodeID *int
 
 	traffic, err := s.GetClientTrafficByEmail(clientEmail)
 	if err != nil {
-		return false, err
+		return false, nil, err
 	}
 
 	if !traffic.Enable {
 		inbound, err := s.GetInbound(id)
 		if err != nil {
-			return false, err
+			return false, nil, err
 		}
 		clients, err := s.GetClients(inbound)
 		if err != nil {
-			return false, err
+			return false, nil, err
 		}
 		for _, client := range clients {
 			if client.Email == clientEmail && client.Enable {
-				rt, push, _, perr := s.nodePushPlan(inbound)
-				if perr != nil {
-					return false, perr
-				}
-				if !push {
-					if inbound.NodeID == nil {
-						needRestart = true
-					}
-					break
-				}
 				cipher := ""
 				if string(inbound.Protocol) == "shadowsocks" {
 					var oldSettings map[string]any
 					err = json.Unmarshal([]byte(inbound.Settings), &oldSettings)
 					if err != nil {
-						return false, err
+						return false, nil, err
 					}
 					cipher, _ = oldSettings["method"].(string)
 				}
-				err1 := rt.AddUser(context.Background(), inbound, map[string]any{
+				clientMap := map[string]any{
 					"email":    client.Email,
 					"id":       client.ID,
 					"auth":     client.Auth,
@@ -634,14 +646,11 @@ func (s *InboundService) resetClientTrafficLocked(id int, clientEmail string) (b
 					"flow":     client.Flow,
 					"password": client.Password,
 					"cipher":   cipher,
-				})
-				if err1 == nil {
-					logger.Debug("Client enabled on", rt.Name(), "due to reset traffic:", clientEmail)
-				} else if inbound.NodeID != nil {
-					logger.Warning("Error in enabling client on", rt.Name(), ":", err1)
+				}
+				if inbound.NodeID != nil {
+					reenableNodeID = inbound.NodeID
 				} else {
-					logger.Debug("Error in enabling client on", rt.Name(), ":", err1)
-					needRestart = true
+					reenablePlan = &trafficLocalApplyPlan{action: trafficAddUser, inbound: *inbound, client: clientMap}
 				}
 				break
 			}
@@ -656,7 +665,7 @@ func (s *InboundService) resetClientTrafficLocked(id int, clientEmail string) (b
 	now := time.Now().UnixMilli()
 	inbound, err := s.GetInbound(id)
 	if err != nil {
-		return false, err
+		return false, nil, err
 	}
 	if err := db.Transaction(func(tx *gorm.DB) error {
 		if err := adjustGroupBaselinesForRemovedTraffic(tx, []string{clientEmail}); err != nil {
@@ -676,25 +685,30 @@ func (s *InboundService) resetClientTrafficLocked(id int, clientEmail string) (b
 			Update("last_traffic_reset_time", now).Error; err != nil {
 			return err
 		}
+		if reenableNodeID != nil {
+			return (&NodeService{}).MarkNodeDirtyTx(tx, *reenableNodeID)
+		}
 		if inbound != nil && inbound.NodeID != nil {
 			return (&NodeService{}).MarkNodeDirtyTx(tx, *inbound.NodeID)
 		}
 		return nil
 	}); err != nil {
-		return false, err
+		return false, nil, err
 	}
 
-	if inbound != nil && inbound.NodeID != nil {
-		if rt, rterr := s.runtimeFor(inbound); rterr == nil {
-			if e := rt.ResetClientTraffic(context.Background(), inbound, clientEmail); e != nil {
-				logger.Warning("ResetClientTraffic: remote propagation to", rt.Name(), "failed:", e)
-			}
+	if reenablePlan != nil {
+		rt, err := s.runtimeFor(&reenablePlan.inbound)
+		if err != nil {
+			needRestart = true
+		} else if err := rt.AddUser(context.Background(), &reenablePlan.inbound, reenablePlan.client); err != nil {
+			logger.Debug("Error in enabling client on", rt.Name(), ":", err)
+			needRestart = true
 		} else {
-			logger.Warning("ResetClientTraffic: runtime lookup failed:", rterr)
+			logger.Debug("Client enabled on", rt.Name(), "due to reset traffic:", clientEmail)
 		}
 	}
 
-	return needRestart, nil
+	return needRestart, inbound, nil
 }
 
 func (s *InboundService) ResetAllTraffics() error {
@@ -740,16 +754,24 @@ func (s *InboundService) propagateResetAllTrafficsToNodes() {
 }
 
 func (s *InboundService) ResetInboundTraffic(id int) error {
+	var inbound *model.Inbound
 	if err := submitTrafficWrite(func() error {
-		return database.GetDB().Model(model.Inbound{}).
+		db := database.GetDB()
+		if err := db.Model(model.Inbound{}).
 			Where("id = ?", id).
-			Updates(map[string]any{"up": 0, "down": 0}).Error
+			Updates(map[string]any{"up": 0, "down": 0}).Error; err != nil {
+			return err
+		}
+		var err error
+		inbound, err = s.GetInbound(id)
+		if err != nil {
+			return err
+		}
+		return nil
 	}); err != nil {
 		return err
 	}
-
-	inbound, err := s.GetInbound(id)
-	if err == nil && inbound != nil && inbound.NodeID != nil {
+	if inbound != nil && inbound.NodeID != nil {
 		if rt, rterr := s.runtimeFor(inbound); rterr == nil {
 			if e := rt.ResetInboundTraffic(context.Background(), inbound); e != nil {
 				logger.Warning("ResetInboundTraffic: remote propagation to", rt.Name(), "failed:", e)
@@ -763,134 +785,161 @@ func (s *InboundService) ResetInboundTraffic(id int) error {
 
 func (s *InboundService) DelDepletedClients(id int) (err error) {
 	db := database.GetDB()
-	tx := db.Begin()
-	defer func() {
-		if err == nil {
-			tx.Commit()
-		} else {
-			tx.Rollback()
+	var deletedInbounds []model.Inbound
+	err = db.Transaction(func(tx *gorm.DB) error {
+		// Collect depleted emails globally — a shared-email row owned by one
+		// inbound depletes every sibling that lists the email.
+		now := time.Now().Unix() * 1000
+		depletedClause := "reset = 0 and ((total > 0 and up + down >= total) or (expiry_time > 0 and expiry_time <= ?))"
+		var depletedRows []xray.ClientTraffic
+		if err := tx.Model(xray.ClientTraffic{}).
+			Where(depletedClause, now).
+			Find(&depletedRows).Error; err != nil {
+			return err
+		}
+		if len(depletedRows) == 0 {
+			return nil
 		}
-	}()
 
-	// Collect depleted emails globally — a shared-email row owned by one
-	// inbound depletes every sibling that lists the email.
-	now := time.Now().Unix() * 1000
-	depletedClause := "reset = 0 and ((total > 0 and up + down >= total) or (expiry_time > 0 and expiry_time <= ?))"
-	var depletedRows []xray.ClientTraffic
-	err = db.Model(xray.ClientTraffic{}).
-		Where(depletedClause, now).
-		Find(&depletedRows).Error
+		depletedEmails := make(map[string]struct{}, len(depletedRows))
+		for _, r := range depletedRows {
+			if r.Email == "" {
+				continue
+			}
+			depletedEmails[strings.ToLower(r.Email)] = struct{}{}
+		}
+		if len(depletedEmails) == 0 {
+			return nil
+		}
+
+		var inbounds []*model.Inbound
+		inboundQuery := tx.Model(model.Inbound{})
+		if id >= 0 {
+			inboundQuery = inboundQuery.Where("id = ?", id)
+		}
+		if err := inboundQuery.Find(&inbounds).Error; err != nil {
+			return err
+		}
+
+		for _, inbound := range inbounds {
+			var settings map[string]any
+			if err := json.Unmarshal([]byte(inbound.Settings), &settings); err != nil {
+				return err
+			}
+			rawClients, ok := settings["clients"].([]any)
+			if !ok {
+				continue
+			}
+			newClients := make([]any, 0, len(rawClients))
+			removed := 0
+			for _, client := range rawClients {
+				c, ok := client.(map[string]any)
+				if !ok {
+					newClients = append(newClients, client)
+					continue
+				}
+				email, _ := c["email"].(string)
+				if _, isDepleted := depletedEmails[strings.ToLower(email)]; isDepleted {
+					removed++
+					continue
+				}
+				newClients = append(newClients, client)
+			}
+			if removed == 0 {
+				continue
+			}
+			if len(newClients) == 0 {
+				deletedInbounds = append(deletedInbounds, *inbound)
+				if err := s.clientService.DetachInbound(tx, inbound.Id); err != nil {
+					return err
+				}
+				if err := tx.Where("inbound_id = ?", inbound.Id).Delete(&model.Host{}).Error; err != nil {
+					return err
+				}
+				if err := tx.Delete(model.Inbound{}, inbound.Id).Error; err != nil {
+					return err
+				}
+				if inbound.NodeID != nil {
+					if err := (&NodeService{}).MarkNodeDirtyTx(tx, *inbound.NodeID); err != nil {
+						return err
+					}
+				}
+				continue
+			}
+			settings["clients"] = newClients
+			ns, mErr := json.MarshalIndent(settings, "", "  ")
+			if mErr != nil {
+				return mErr
+			}
+			inbound.Settings = string(ns)
+			if err := tx.Save(inbound).Error; err != nil {
+				return err
+			}
+			survivingClients, gcErr := s.GetClients(inbound)
+			if gcErr != nil {
+				return gcErr
+			}
+			if err := s.clientService.SyncInbound(tx, inbound.Id, survivingClients); err != nil {
+				return err
+			}
+			if inbound.NodeID != nil {
+				if err := (&NodeService{}).MarkNodeDirtyTx(tx, *inbound.NodeID); err != nil {
+					return err
+				}
+			}
+		}
+
+		// Drop now-orphaned rows. With id >= 0, a row is safe to drop only when
+		// no out-of-scope inbound still references the email.
+		if id < 0 {
+			return tx.Where(depletedClause, now).Delete(xray.ClientTraffic{}).Error
+		}
+		emails := make([]string, 0, len(depletedEmails))
+		for e := range depletedEmails {
+			emails = append(emails, e)
+		}
+		var stillReferenced []string
+		emailExpr := database.JSONFieldText("client.value", "email")
+		stillQuery := fmt.Sprintf(
+			"SELECT DISTINCT LOWER(%s) %s WHERE LOWER(%s) IN ?",
+			emailExpr,
+			database.JSONClientsFromInbound(),
+			emailExpr,
+		)
+		if err := tx.Raw(stillQuery, emails).Scan(&stillReferenced).Error; err != nil {
+			return err
+		}
+		stillSet := make(map[string]struct{}, len(stillReferenced))
+		for _, e := range stillReferenced {
+			stillSet[e] = struct{}{}
+		}
+		toDelete := make([]string, 0, len(emails))
+		for _, e := range emails {
+			if _, kept := stillSet[e]; !kept {
+				toDelete = append(toDelete, e)
+			}
+		}
+		if len(toDelete) > 0 {
+			if err := tx.Where("LOWER(email) IN ?", toDelete).Delete(xray.ClientTraffic{}).Error; err != nil {
+				return err
+			}
+		}
+		return nil
+	})
 	if err != nil {
 		return err
 	}
-	if len(depletedRows) == 0 {
-		return nil
-	}
-
-	depletedEmails := make(map[string]struct{}, len(depletedRows))
-	for _, r := range depletedRows {
-		if r.Email == "" {
-			continue
+	for i := range deletedInbounds {
+		inbound := &deletedInbounds[i]
+		if rt, rtErr := s.runtimeFor(inbound); rtErr != nil {
+			logger.Warning("DelDepletedClients: runtime lookup failed after commit:", rtErr)
+		} else if rtErr = rt.DelInbound(context.Background(), inbound); rtErr != nil && !xray.IsMissingHandlerErr(rtErr) {
+			logger.Warning("DelDepletedClients: runtime cleanup failed after commit:", rtErr)
 		}
-		depletedEmails[strings.ToLower(r.Email)] = struct{}{}
-	}
-	if len(depletedEmails) == 0 {
-		return nil
-	}
-
-	var inbounds []*model.Inbound
-	inboundQuery := db.Model(model.Inbound{})
-	if id >= 0 {
-		inboundQuery = inboundQuery.Where("id = ?", id)
-	}
-	if err = inboundQuery.Find(&inbounds).Error; err != nil {
-		return err
-	}
-
-	for _, inbound := range inbounds {
-		var settings map[string]any
-		if err = json.Unmarshal([]byte(inbound.Settings), &settings); err != nil {
-			return err
-		}
-		rawClients, ok := settings["clients"].([]any)
-		if !ok {
-			continue
-		}
-		newClients := make([]any, 0, len(rawClients))
-		removed := 0
-		for _, client := range rawClients {
-			c, ok := client.(map[string]any)
-			if !ok {
-				newClients = append(newClients, client)
-				continue
+		if inbound.Tag != "" {
+			if _, syncErr := (&XraySettingService{}).RemoveInboundTagReferences(inbound.Tag); syncErr != nil {
+				logger.Warning("DelDepletedClients: routing cleanup failed after commit:", syncErr)
 			}
-			email, _ := c["email"].(string)
-			if _, isDepleted := depletedEmails[strings.ToLower(email)]; isDepleted {
-				removed++
-				continue
-			}
-			newClients = append(newClients, client)
-		}
-		if removed == 0 {
-			continue
-		}
-		if len(newClients) == 0 {
-			_, _ = s.DelInbound(inbound.Id)
-			continue
-		}
-		settings["clients"] = newClients
-		ns, mErr := json.MarshalIndent(settings, "", "  ")
-		if mErr != nil {
-			return mErr
-		}
-		inbound.Settings = string(ns)
-		if err = tx.Save(inbound).Error; err != nil {
-			return err
-		}
-		survivingClients, gcErr := s.GetClients(inbound)
-		if gcErr != nil {
-			err = gcErr
-			return err
-		}
-		if err = s.clientService.SyncInbound(tx, inbound.Id, survivingClients); err != nil {
-			return err
-		}
-	}
-
-	// Drop now-orphaned rows. With id >= 0, a row is safe to drop only when
-	// no out-of-scope inbound still references the email.
-	if id < 0 {
-		err = tx.Where(depletedClause, now).Delete(xray.ClientTraffic{}).Error
-		return err
-	}
-	emails := make([]string, 0, len(depletedEmails))
-	for e := range depletedEmails {
-		emails = append(emails, e)
-	}
-	var stillReferenced []string
-	emailExpr := database.JSONFieldText("client.value", "email")
-	stillQuery := fmt.Sprintf(
-		"SELECT DISTINCT LOWER(%s) %s WHERE LOWER(%s) IN ?",
-		emailExpr,
-		database.JSONClientsFromInbound(),
-		emailExpr,
-	)
-	if err = tx.Raw(stillQuery, emails).Scan(&stillReferenced).Error; err != nil {
-		return err
-	}
-	stillSet := make(map[string]struct{}, len(stillReferenced))
-	for _, e := range stillReferenced {
-		stillSet[e] = struct{}{}
-	}
-	toDelete := make([]string, 0, len(emails))
-	for _, e := range emails {
-		if _, kept := stillSet[e]; !kept {
-			toDelete = append(toDelete, e)
-		}
-	}
-	if len(toDelete) > 0 {
-		if err = tx.Where("LOWER(email) IN ?", toDelete).Delete(xray.ClientTraffic{}).Error; err != nil {
-			return err
 		}
 	}
 	return nil

internal/web/service/inbound_traffic_apply.go

@@ -0,0 +1,105 @@
+package service
+
+import (
+	"context"
+	"strings"
+
+	"github.com/mhsanaei/3x-ui/v3/internal/database/model"
+	"github.com/mhsanaei/3x-ui/v3/internal/logger"
+	"github.com/mhsanaei/3x-ui/v3/internal/xray"
+
+	"gorm.io/gorm"
+)
+
+type trafficLocalApplyAction uint8
+
+const (
+	trafficAddUser trafficLocalApplyAction = iota + 1
+	trafficRemoveUser
+	trafficDisableInbound
+)
+
+type trafficLocalApplyPlan struct {
+	action  trafficLocalApplyAction
+	inbound model.Inbound
+	client  map[string]any
+	email   string
+}
+
+type trafficMutationBatch struct {
+	localPlans  []trafficLocalApplyPlan
+	remotePlans []trafficInboundUpdatePlan
+	nodeIDs     map[int]struct{}
+}
+
+type trafficInboundUpdatePlan struct{ oldInbound, newInbound model.Inbound }
+
+func newTrafficMutationBatch() *trafficMutationBatch {
+	return &trafficMutationBatch{nodeIDs: make(map[int]struct{})}
+}
+
+func (b *trafficMutationBatch) addNode(nodeID int) {
+	if nodeID > 0 {
+		b.nodeIDs[nodeID] = struct{}{}
+	}
+}
+
+func (b *trafficMutationBatch) markNodesTx(tx *gorm.DB) error {
+	if b == nil {
+		return nil
+	}
+	nodeSvc := NodeService{}
+	for nodeID := range b.nodeIDs {
+		if err := nodeSvc.MarkNodeDirtyTx(tx, nodeID); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (s *InboundService) applyTrafficMutationBatch(b *trafficMutationBatch) bool {
+	if b == nil {
+		return false
+	}
+	needRestart := false
+	for i := range b.remotePlans {
+		plan := &b.remotePlans[i]
+		rt, err := s.runtimeFor(&plan.newInbound)
+		if err == nil {
+			err = rt.UpdateInbound(context.Background(), &plan.oldInbound, &plan.newInbound)
+		}
+		if err != nil {
+			logger.Debug("traffic post-commit remote apply failed:", err)
+			needRestart = true
+		}
+	}
+	for i := range b.localPlans {
+		plan := &b.localPlans[i]
+		if plan.inbound.Protocol == model.MTProto {
+			s.applyLocalMtproto(plan.inbound.Id)
+			continue
+		}
+		rt, err := s.runtimeFor(&plan.inbound)
+		if err == nil {
+			switch plan.action {
+			case trafficAddUser:
+				err = rt.AddUser(context.Background(), &plan.inbound, plan.client)
+			case trafficRemoveUser:
+				err = rt.RemoveUser(context.Background(), &plan.inbound, plan.email)
+				if err != nil && strings.Contains(err.Error(), "not found") {
+					err = nil
+				}
+			case trafficDisableInbound:
+				err = rt.DelInbound(context.Background(), &plan.inbound)
+				if xray.IsMissingHandlerErr(err) {
+					err = nil
+				}
+			}
+		}
+		if err != nil {
+			logger.Debug("traffic post-commit runtime apply failed:", err)
+			needRestart = true
+		}
+	}
+	return needRestart
+}

internal/web/service/traffic_runtime_apply_test.go

@@ -0,0 +1,86 @@
+package service
+
+import (
+	"testing"
+
+	"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"
+)
+
+func TestTrafficDisableImmediatelyUpdatesNodeRuntime(t *testing.T) {
+	setupConflictDB(t)
+	nodeID, fake := setupNodeRuntime(t)
+	client := model.Client{Email: "spent-node", Enable: true}
+	ib := nodeInbound(t, nodeID, 46301, []model.Client{client})
+	if err := database.GetDB().Create(&xray.ClientTraffic{
+		InboundId: ib.Id, Email: client.Email, Enable: true, Up: 100, Total: 100,
+	}).Error; err != nil {
+		t.Fatalf("seed traffic: %v", err)
+	}
+
+	if _, _, _, err := (&InboundService{}).addTrafficLocked(nil, nil); err != nil {
+		t.Fatalf("addTrafficLocked: %v", err)
+	}
+	if got := fake.updateInbound.Load(); got != 1 {
+		t.Fatalf("remote UpdateInbound calls = %d, want 1 after commit", got)
+	}
+}
+
+func TestTrafficDisableRefreshesLocalMTProtoSidecar(t *testing.T) {
+	setupConflictDB(t)
+	mgr := runtime.NewManager(runtime.LocalDeps{APIPort: func() int { return 0 }})
+	fake := &fakeNodeRuntime{}
+	mgr.SetLocalRuntimeOverride(fake)
+	runtime.SetManager(mgr)
+	t.Cleanup(func() { runtime.SetManager(nil) })
+
+	seedInboundConflict(t, "mt-spent", "", 46302, model.MTProto, "",
+		`{"clients":[{"email":"spent-mt","secret":"`+mtprotoTestSecretA+`","enable":true}]}`)
+	ib := loadInboundByTag(t, "mt-spent")
+	clients, err := (&InboundService{}).GetClients(ib)
+	if err != nil {
+		t.Fatalf("GetClients: %v", err)
+	}
+	if err := (&ClientService{}).SyncInbound(nil, ib.Id, clients); err != nil {
+		t.Fatalf("SyncInbound: %v", err)
+	}
+	seedClientTraffic(t, ib.Id, "spent-mt", true)
+	if err := database.GetDB().Model(&xray.ClientTraffic{}).Where("email = ?", "spent-mt").
+		Updates(map[string]any{"up": 100, "total": 100}).Error; err != nil {
+		t.Fatalf("deplete traffic: %v", err)
+	}
+
+	if _, _, _, err := (&InboundService{}).addTrafficLocked(nil, nil); err != nil {
+		t.Fatalf("addTrafficLocked: %v", err)
+	}
+	if got := fake.updateInbound.Load(); got != 1 {
+		t.Fatalf("MTProto sidecar UpdateInbound calls = %d, want 1 after commit", got)
+	}
+}
+
+func TestDelDepletedClientsCleansRuntimeAfterCommit(t *testing.T) {
+	setupConflictDB(t)
+	mgr := runtime.NewManager(runtime.LocalDeps{APIPort: func() int { return 0 }})
+	fake := &fakeNodeRuntime{}
+	mgr.SetLocalRuntimeOverride(fake)
+	runtime.SetManager(mgr)
+	t.Cleanup(func() { runtime.SetManager(nil) })
+
+	seedInboundConflict(t, "depleted-only", "", 46303, model.VLESS, `{"network":"tcp"}`,
+		`{"clients":[{"email":"gone","enable":true}]}`)
+	ib := loadInboundByTag(t, "depleted-only")
+	seedClientTraffic(t, ib.Id, "gone", true)
+	if err := database.GetDB().Model(&xray.ClientTraffic{}).Where("email = ?", "gone").
+		Updates(map[string]any{"up": 100, "total": 100, "reset": 0}).Error; err != nil {
+		t.Fatalf("deplete traffic: %v", err)
+	}
+
+	if err := (&InboundService{}).DelDepletedClients(-1); err != nil {
+		t.Fatalf("DelDepletedClients: %v", err)
+	}
+	if got := fake.delInbound.Load(); got != 1 {
+		t.Fatalf("runtime DelInbound calls = %d, want 1 after commit", got)
+	}
+}