diff --git a/frontend/src/generated/types.ts b/frontend/src/generated/types.ts index acfdcad7c..d5545d127 100644 --- a/frontend/src/generated/types.ts +++ b/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 { diff --git a/frontend/src/generated/zod.ts b/frontend/src/generated/zod.ts index d4466c239..5ef58fb2c 100644 --- a/frontend/src/generated/zod.ts +++ b/frontend/src/generated/zod.ts @@ -15,6 +15,9 @@ export type SubLinkProvider = z.infer; export const staticEgressResolverSchema = z.string(); export type staticEgressResolver = z.infer; +export const trafficLocalApplyActionSchema = z.number().int(); +export type trafficLocalApplyAction = z.infer; + export const transportBitsSchema = z.number().int(); export type transportBits = z.infer; diff --git a/internal/web/service/global_traffic_test.go b/internal/web/service/global_traffic_test.go index 34fb1b3e4..c4ce99c9b 100644 --- a/internal/web/service/global_traffic_test.go +++ b/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) diff --git a/internal/web/service/inbound.go b/internal/web/service/inbound.go index d4d975196..09753b7b7 100644 --- a/internal/web/service/inbound.go +++ b/internal/web/service/inbound.go @@ -29,7 +29,6 @@ import ( ) type InboundService struct { - xrayApi xray.XrayAPI clientService ClientService fallbackService FallbackService } diff --git a/internal/web/service/inbound_autorenew_shadowsocks_test.go b/internal/web/service/inbound_autorenew_shadowsocks_test.go index 7cd7339f6..228e59f30 100644 --- a/internal/web/service/inbound_autorenew_shadowsocks_test.go +++ b/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) diff --git a/internal/web/service/inbound_autorenew_test.go b/internal/web/service/inbound_autorenew_test.go index 34f6ba0ed..54a7b6840 100644 --- a/internal/web/service/inbound_autorenew_test.go +++ b/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) diff --git a/internal/web/service/inbound_disable.go b/internal/web/service/inbound_disable.go index 01b8f7df9..a4e6bec58 100644 --- a/internal/web/service/inbound_disable.go +++ b/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 -} diff --git a/internal/web/service/inbound_node.go b/internal/web/service/inbound_node.go index 9f69a44d3..ef4c4adb9 100644 --- a/internal/web/service/inbound_node.go +++ b/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 { diff --git a/internal/web/service/inbound_traffic.go b/internal/web/service/inbound_traffic.go index ae47a8ad4..339329b4d 100644 --- a/internal/web/service/inbound_traffic.go +++ b/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 diff --git a/internal/web/service/inbound_traffic_apply.go b/internal/web/service/inbound_traffic_apply.go new file mode 100644 index 000000000..d7876e20f --- /dev/null +++ b/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 +} diff --git a/internal/web/service/traffic_runtime_apply_test.go b/internal/web/service/traffic_runtime_apply_test.go new file mode 100644 index 000000000..f038d4180 --- /dev/null +++ b/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) + } +}