diff --git a/internal/web/service/client_crud.go b/internal/web/service/client_crud.go index 30a8cc640..1f2d1a761 100644 --- a/internal/web/service/client_crud.go +++ b/internal/web/service/client_crud.go @@ -782,6 +782,12 @@ func (s *ClientService) update(inboundSvc *InboundService, id int, updated model updated.Secret = existing.Secret } + // A client the quota loop switched off comes back on when this update + // restores a usable window, as BulkAdjust already does on a renewal (#5619). + if !updated.Enable && s.renewalReenables(existing.Email, updated.TotalGB, updated.ExpiryTime) { + updated.Enable = true + } + if updated.Email != existing.Email { var collisionCount int64 if err := database.GetDB().Model(&model.ClientRecord{}). @@ -936,6 +942,15 @@ func (s *ClientService) update(inboundSvc *InboundService, id int, updated model }).Error; err != nil { return needRestart, err } + // No inbound left, so UpdateClientStat never runs; re-assert the + // enforcement row's off switch here or a re-enabled client stays off. + if merged.Enable { + if err := database.GetDB().Model(xray.ClientTraffic{}). + Where("email = ?", existing.Email). + UpdateColumn("enable", true).Error; err != nil { + return needRestart, err + } + } } reverseStr := "" diff --git a/internal/web/service/client_inbound_apply.go b/internal/web/service/client_inbound_apply.go index c9df9de98..f0c9c1882 100644 --- a/internal/web/service/client_inbound_apply.go +++ b/internal/web/service/client_inbound_apply.go @@ -1537,9 +1537,31 @@ func (s *ClientService) ResetClientIpLimitByEmail(inboundSvc *InboundService, cl }) } +// renewalReenableFor projects the post-renewal limits for a single-field +// renewal (nil side keeps the stored limit; no row means no quota state). +func (s *ClientService) renewalReenableFor(email string, setExpiry *int64, setTotal *int64) bool { + total, expiry, ok := storedTrafficLimits(email) + if !ok { + return false + } + if setExpiry != nil { + expiry = *setExpiry + } + if setTotal != nil { + total = *setTotal + } + return s.renewalReenables(email, total, expiry) +} + func (s *ClientService) ResetClientExpiryTimeByEmail(inboundSvc *InboundService, clientEmail string, expiry_time int64) (bool, error) { + reenable := s.renewalReenableFor(clientEmail, &expiry_time, nil) return s.applyClientFieldByEmail(inboundSvc, clientEmail, func(c map[string]any) { c["expiryTime"] = expiry_time + if reenable { + // A renewed client that the quota loop had switched off comes back + // with the renewal instead of waiting for a manual enable (#5619). + c["enable"] = true + } }) } @@ -1547,7 +1569,12 @@ func (s *ClientService) ResetClientTrafficLimitByEmail(inboundSvc *InboundServic if totalGB < 0 { return false, common.NewError("totalGB must be >= 0") } + newTotal := int64(totalGB) * 1024 * 1024 * 1024 + reenable := s.renewalReenableFor(clientEmail, nil, &newTotal) return s.applyClientFieldByEmail(inboundSvc, clientEmail, func(c map[string]any) { - c["totalGB"] = totalGB * 1024 * 1024 * 1024 + c["totalGB"] = newTotal + if reenable { + c["enable"] = true + } }) } diff --git a/internal/web/service/client_renewal_reenable.go b/internal/web/service/client_renewal_reenable.go new file mode 100644 index 000000000..edd9a9deb --- /dev/null +++ b/internal/web/service/client_renewal_reenable.go @@ -0,0 +1,60 @@ +package service + +import ( + "time" + + "github.com/mhsanaei/3x-ui/v3/internal/database" + "github.com/mhsanaei/3x-ui/v3/internal/database/model" + "github.com/mhsanaei/3x-ui/v3/internal/xray" +) + +// renewalReenables reports whether the quota loop switched the client off and +// this renewal (newTotal/newExpiry, usage unchanged) clears the depletion (#5619). +func (s *ClientService) renewalReenables(email string, newTotal, newExpiry int64) bool { + db := database.GetDB() + var traffic xray.ClientTraffic + if err := db.Where("email = ?", email).First(&traffic).Error; err != nil { + return false // no enforcement row: the quota loop never switched this client off + } + if traffic.Enable { + return false + } + + cond, condArgs := depletedCond(db) + var wasDepleted int64 + if err := db.Model(xray.ClientTraffic{}). + Where(cond+" AND email = ?", append(append([]any{}, condArgs...), email)...). + Count(&wasDepleted).Error; err != nil || wasDepleted == 0 { + return false // off for another reason than depletion: the operator's off stands + } + + // Project the post-renewal row: total/expiry move, up/down and a fresh + // cross-panel push can still deplete it. + now := time.Now().UnixMilli() + used := traffic.Up + traffic.Down + if newTotal > 0 && used >= newTotal { + return false // still over quota + } + if newExpiry > 0 && newExpiry <= now { + return false // still expired + } + if newTotal > 0 { + var globalDepleted int64 + if err := db.Model(&model.ClientGlobalTraffic{}). + Where("email = ? AND updated_at >= ? AND up + down >= ?", email, globalTrafficFreshSince(time.Now()), newTotal). + Count(&globalDepleted).Error; err == nil && globalDepleted > 0 { + return false // a master's pushed usage still depletes the fresh quota + } + } + return true +} + +// storedTrafficLimits reads a client's enforcement row; ok is false when the +// client has no row, in which case no quota state can be projected for it. +func storedTrafficLimits(email string) (total, expiry int64, ok bool) { + var row xray.ClientTraffic + if err := database.GetDB().Where("email = ?", email).First(&row).Error; err != nil { + return 0, 0, false + } + return row.Total, row.ExpiryTime, true +} diff --git a/internal/web/service/client_traffic.go b/internal/web/service/client_traffic.go index 73d663d2e..025c83c39 100644 --- a/internal/web/service/client_traffic.go +++ b/internal/web/service/client_traffic.go @@ -50,17 +50,22 @@ func (s *ClientService) ResetTrafficByEmail(inboundSvc *InboundService, email st needRestart = nr } - // Enable only a client the quota switched off, once the counters are zero: a - // hand-disabled one stays off, and a still-depleted one would be switched off again. + // Restore quota-disabled clients after zeroing; preserve an operator + // disable, including when no inbound remains. if !rec.Enable && len(quotaDisabled) > 0 { - updated := rec.ToClient() - updated.Enable = true - nr, uErr := s.Update(inboundSvc, rec.Id, *updated, rec.LimitHwid) - if uErr != nil { - logger.Warning("Failed to auto-enable client during traffic reset:", uErr) - } - if nr { - needRestart = true + // Inbound resets already restored hosted clients. Only an unhosted + // record still needs the full update. + fresh, fErr := s.GetRecordByEmail(nil, email) + if fErr == nil && !fresh.Enable { + updated := fresh.ToClient() + updated.Enable = true + nr, uErr := s.Update(inboundSvc, fresh.Id, *updated, fresh.LimitHwid) + if uErr != nil { + logger.Warning("Failed to auto-enable client during traffic reset:", uErr) + } + if nr { + needRestart = true + } } } return needRestart, nil diff --git a/internal/web/service/client_traffic_reset_order_test.go b/internal/web/service/client_traffic_reset_order_test.go index 152c0738e..f69ba75ea 100644 --- a/internal/web/service/client_traffic_reset_order_test.go +++ b/internal/web/service/client_traffic_reset_order_test.go @@ -67,7 +67,7 @@ func requireResetSurvivesTick(t *testing.T, reset func() error) { tickDone <- err }() deadline := time.Now().Add(5 * time.Second) - for len(twQueue) < 2 && time.Now().Before(deadline) { + for trafficWriterQueueLen() < 2 && time.Now().Before(deadline) { time.Sleep(time.Millisecond) } close(release) diff --git a/internal/web/service/client_update_reenable_test.go b/internal/web/service/client_update_reenable_test.go new file mode 100644 index 000000000..28dfa5c24 --- /dev/null +++ b/internal/web/service/client_update_reenable_test.go @@ -0,0 +1,284 @@ +package service + +import ( + "testing" + "time" + + "github.com/mhsanaei/3x-ui/v3/internal/database" + "github.com/mhsanaei/3x-ui/v3/internal/database/model" + "github.com/mhsanaei/3x-ui/v3/internal/xray" +) + +// renewalReenableDay keeps the seeds one day past (or before) their boundary. +const renewalReenableDay = int64(24 * 60 * 60 * 1000) + +// seedRenewableDepletedClient seeds a local inbound whose client the quota loop +// switched off: record off, inbound JSON off, and the traffic row off and depleted. +func seedRenewableDepletedClient(t *testing.T, svc *ClientService, port int, email string, total, expiry, up, down int64) *model.Inbound { + t.Helper() + ib := seedLocalDisabledClient(t, svc, port, "", email, total, expiry, up, down) + if got := trafficOf(t, email); got.Enable { + t.Fatalf("seed %s: want an off traffic row, got %+v", email, got) + } + return ib +} + +func TestUpdate_ReenablesDepletedClient_WhenExpiryExtended(t *testing.T) { + setupBulkDB(t) + svc := &ClientService{} + inboundSvc := &InboundService{} + + now := time.Now().UnixMilli() + email := "renew-exp@x" + ib := seedRenewableDepletedClient(t, svc, 53101, email, 0, now-2*renewalReenableDay, 0, 0) + + rec, err := svc.GetRecordByEmail(nil, email) + if err != nil { + t.Fatalf("GetRecordByEmail: %v", err) + } + updated := rec.ToClient() + updated.ExpiryTime = now + 30*renewalReenableDay + // The editor round-trips the switch as the stored off state; the renewal + // must turn it back on by itself. + if _, err := svc.Update(inboundSvc, rec.Id, *updated, 0); err != nil { + t.Fatalf("Update: %v", err) + } + assertEnableEverywhere(t, svc, inboundSvc, ib.Id, email, true) + if got := trafficOf(t, email).ExpiryTime; got != now+30*renewalReenableDay { + t.Fatalf("%s: expiry = %d, want %d", email, got, now+30*renewalReenableDay) + } +} + +func TestUpdate_ReenablesDepletedClient_WhenTotalRaised(t *testing.T) { + setupBulkDB(t) + svc := &ClientService{} + inboundSvc := &InboundService{} + + email := "renew-total@x" + ib := seedRenewableDepletedClient(t, svc, 53102, email, 100, 0, 60, 40) + + rec, err := svc.GetRecordByEmail(nil, email) + if err != nil { + t.Fatalf("GetRecordByEmail: %v", err) + } + updated := rec.ToClient() + updated.TotalGB = 300 + if _, err := svc.Update(inboundSvc, rec.Id, *updated, 0); err != nil { + t.Fatalf("Update: %v", err) + } + assertEnableEverywhere(t, svc, inboundSvc, ib.Id, email, true) + if got := trafficOf(t, email).Total; got != 300 { + t.Fatalf("%s: total = %d, want 300", email, got) + } +} + +func TestUpdate_KeepsDisabled_StillExpiredAfterUpdate(t *testing.T) { + setupBulkDB(t) + svc := &ClientService{} + inboundSvc := &InboundService{} + + now := time.Now().UnixMilli() + email := "renew-sml@x" + ib := seedRenewableDepletedClient(t, svc, 53103, email, 0, now-10*renewalReenableDay, 0, 0) + + rec, err := svc.GetRecordByEmail(nil, email) + if err != nil { + t.Fatalf("GetRecordByEmail: %v", err) + } + updated := rec.ToClient() + updated.ExpiryTime = now - 5*renewalReenableDay // an extension that still lands in the past + if _, err := svc.Update(inboundSvc, rec.Id, *updated, 0); err != nil { + t.Fatalf("Update: %v", err) + } + assertEnableEverywhere(t, svc, inboundSvc, ib.Id, email, false) +} + +func TestUpdate_KeepsDisabled_StillOverQuotaAfterUpdate(t *testing.T) { + setupBulkDB(t) + svc := &ClientService{} + inboundSvc := &InboundService{} + + email := "renew-quota@x" + ib := seedRenewableDepletedClient(t, svc, 53104, email, 100, 0, 60, 40) + + rec, err := svc.GetRecordByEmail(nil, email) + if err != nil { + t.Fatalf("GetRecordByEmail: %v", err) + } + updated := rec.ToClient() + updated.TotalGB = 80 // still below the 100 bytes of usage + if _, err := svc.Update(inboundSvc, rec.Id, *updated, 0); err != nil { + t.Fatalf("Update: %v", err) + } + assertEnableEverywhere(t, svc, inboundSvc, ib.Id, email, false) +} + +func TestUpdate_KeepsDisabled_ManuallyDisabledNotDepleted(t *testing.T) { + setupBulkDB(t) + svc := &ClientService{} + inboundSvc := &InboundService{} + + now := time.Now().UnixMilli() + email := "renew-manual@x" + ib := seedRenewableDepletedClient(t, svc, 53105, email, 0, now+30*renewalReenableDay, 0, 0) + + rec, err := svc.GetRecordByEmail(nil, email) + if err != nil { + t.Fatalf("GetRecordByEmail: %v", err) + } + updated := rec.ToClient() + updated.ExpiryTime = now + 60*renewalReenableDay + if _, err := svc.Update(inboundSvc, rec.Id, *updated, 0); err != nil { + t.Fatalf("Update: %v", err) + } + // The quota loop never switched this client off — it reads under quota — + // so a manual switch-off survives a renewal. + assertEnableEverywhere(t, svc, inboundSvc, ib.Id, email, false) +} + +func TestUpdate_ReenablesDepletedClient_SkipsStillDepletedGlobalUsage(t *testing.T) { + setupBulkDB(t) + svc := &ClientService{} + inboundSvc := &InboundService{} + + now := time.Now().UnixMilli() + email := "renew-global@x" + ib := seedRenewableDepletedClient(t, svc, 53106, email, 100, now+30*renewalReenableDay, 0, 0) + // A fresh master push depletes the local limits even though up/down read 0. + global := model.ClientGlobalTraffic{MasterGuid: "master-a", Email: email, Up: 90, Down: 60} + if err := database.GetDB().Create(&global).Error; err != nil { + t.Fatalf("seed global traffic: %v", err) + } + + rec, err := svc.GetRecordByEmail(nil, email) + if err != nil { + t.Fatalf("GetRecordByEmail: %v", err) + } + updated := rec.ToClient() + updated.TotalGB = 100 + if _, err := svc.Update(inboundSvc, rec.Id, *updated, 0); err != nil { + t.Fatalf("Update: %v", err) + } + assertEnableEverywhere(t, svc, inboundSvc, ib.Id, email, false) + + // Raising the quota above the pushed usage re-enables it. + updated.TotalGB = 1000 + if _, err := svc.Update(inboundSvc, rec.Id, *updated, 0); err != nil { + t.Fatalf("Update: %v", err) + } + assertEnableEverywhere(t, svc, inboundSvc, ib.Id, email, true) +} + +func TestUpdate_ReenablesDepletedClient_WithoutInbound(t *testing.T) { + setupBulkDB(t) + svc := &ClientService{} + inboundSvc := &InboundService{} + + now := time.Now().UnixMilli() + email := "renew-noib@x" + rec := &model.ClientRecord{ + Email: email, + UUID: "44444444-4444-4444-4444-444444444444", + SubID: email, + } + if err := database.GetDB().Create(rec).Error; err != nil { + t.Fatalf("create record: %v", err) + } + forceRecordDisabled(t, svc, email) + if err := database.GetDB().Create(&xray.ClientTraffic{ + Email: email, Enable: false, ExpiryTime: now - 2*renewalReenableDay, + }).Error; err != nil { + t.Fatalf("seed client_traffics: %v", err) + } + + updated := rec.ToClient() + updated.ExpiryTime = now + 30*renewalReenableDay + if _, err := svc.Update(inboundSvc, rec.Id, *updated, 0); err != nil { + t.Fatalf("Update: %v", err) + } + if got := recordEnableOf(t, svc, email); !got { + t.Fatalf("%s: client_records.enable = false, want true", email) + } + if got := trafficOf(t, email).Enable; !got { + t.Fatalf("%s: client_traffics.enable = false, want true (no-inbound re-enable gap)", email) + } +} + +func TestResetClientExpiryTimeByEmail_ReenablesDepletedClient(t *testing.T) { + setupBulkDB(t) + svc := &ClientService{} + inboundSvc := &InboundService{} + + now := time.Now().UnixMilli() + email := "renew-setter@x" + ib := seedRenewableDepletedClient(t, svc, 53107, email, 0, now-2*renewalReenableDay, 0, 0) + + if _, err := svc.ResetClientExpiryTimeByEmail(inboundSvc, email, now+30*renewalReenableDay); err != nil { + t.Fatalf("ResetClientExpiryTimeByEmail: %v", err) + } + assertEnableEverywhere(t, svc, inboundSvc, ib.Id, email, true) +} + +func TestResetClientExpiryTimeByEmail_KeepsDisabled_ManuallyDisabledNotDepleted(t *testing.T) { + setupBulkDB(t) + svc := &ClientService{} + inboundSvc := &InboundService{} + + now := time.Now().UnixMilli() + email := "renew-setter-off@x" + ib := seedRenewableDepletedClient(t, svc, 53108, email, 0, now+30*renewalReenableDay, 0, 0) + + if _, err := svc.ResetClientExpiryTimeByEmail(inboundSvc, email, now+60*renewalReenableDay); err != nil { + t.Fatalf("ResetClientExpiryTimeByEmail: %v", err) + } + assertEnableEverywhere(t, svc, inboundSvc, ib.Id, email, false) +} + +func TestResetClientTrafficLimitByEmail_ReenablesDepletedClient(t *testing.T) { + setupBulkDB(t) + svc := &ClientService{} + inboundSvc := &InboundService{} + + email := "renew-setter-total@x" + ib := seedRenewableDepletedClient(t, svc, 53109, email, 100, 0, 60, 40) + + if _, err := svc.ResetClientTrafficLimitByEmail(inboundSvc, email, 1); err != nil { + t.Fatalf("ResetClientTrafficLimitByEmail: %v", err) + } + assertEnableEverywhere(t, svc, inboundSvc, ib.Id, email, true) + if got := trafficOf(t, email).Total; got != 1<<30 { + t.Fatalf("%s: total = %d, want %d", email, got, 1<<30) + } +} + +// A client living on several inbounds must come back on every one of them, +// not only the inbound whose reset ran first. +func TestResetTrafficByEmail_ReenablesEveryInboundOfTheClient(t *testing.T) { + setupBulkDB(t) + svc := &ClientService{} + inboundSvc := &InboundService{} + + email := "renew-multi@x" + c := model.Client{ + Email: email, ID: "11111111-1111-1111-1111-111111111111", SubID: email, + Enable: false, TotalGB: 1000, + } + ibA := mkInbound(t, 53201, model.VLESS, clientsSettings(t, []model.Client{c})) + ibB := mkInbound(t, 53202, model.VLESS, clientsSettings(t, []model.Client{c})) + for _, ib := range []*model.Inbound{ibA, ibB} { + if err := svc.SyncInbound(nil, ib.Id, []model.Client{c}); err != nil { + t.Fatalf("SyncInbound %d: %v", ib.Id, err) + } + } + mkTraffic(t, ibA.Id, email, 600, 500, 1000, 0, false) + forceRecordDisabled(t, svc, email) + + if _, err := svc.ResetTrafficByEmail(inboundSvc, email); err != nil { + t.Fatalf("ResetTrafficByEmail: %v", err) + } + assertEnableEverywhere(t, svc, inboundSvc, ibA.Id, email, true) + assertEnableEverywhere(t, svc, inboundSvc, ibB.Id, email, true) + if got := trafficOf(t, email); !got.Enable || got.Up+got.Down != 0 { + t.Fatalf("%s: traffic row enable=%v used=%d, want enabled at 0", email, got.Enable, got.Up+got.Down) + } +} diff --git a/internal/web/service/inbound_disable.go b/internal/web/service/inbound_disable.go index 597b8853b..32ad298dd 100644 --- a/internal/web/service/inbound_disable.go +++ b/internal/web/service/inbound_disable.go @@ -275,6 +275,48 @@ func (s *InboundService) markClientsDisabledInSettings(tx *gorm.DB, inboundID in return &snapshot, &ib, nil } +// markClientEnabledInSettings is the re-enable twin of +// markClientsDisabledInSettings: flip one email's stored entry back on. +func (s *InboundService) markClientEnabledInSettings(tx *gorm.DB, inboundID int, email string) error { + var ib model.Inbound + if err := tx.Model(&model.Inbound{}).Where("id = ?", inboundID).First(&ib).Error; err != nil { + return err + } + settings := map[string]any{} + if err := json.Unmarshal([]byte(ib.Settings), &settings); err != nil { + return err + } + clients, _ := settings["clients"].([]any) + now := time.Now().Unix() * 1000 + mutated := false + for i := range clients { + entry, ok := clients[i].(map[string]any) + if !ok { + continue + } + if e, _ := entry["email"].(string); e != email { + continue + } + if cur, _ := entry["enable"].(bool); cur { + continue + } + entry["enable"] = true + entry["updated_at"] = now + clients[i] = entry + mutated = true + } + if !mutated { + return nil + } + settings["clients"] = clients + bs, err := json.MarshalIndent(settings, "", " ") + if err != nil { + return err + } + return tx.Model(&model.Inbound{}).Where("id = ?", inboundID). + Update("settings", string(bs)).Error +} + // trafficDisabledEmails reports which emails have a switched-off stats row. The // table is email-keyed and its inbound_id goes stale, so never filter on it. func trafficDisabledEmails(db *gorm.DB, emails []string) (map[string]struct{}, error) { diff --git a/internal/web/service/inbound_node.go b/internal/web/service/inbound_node.go index f14ce8987..0cf2ad2f4 100644 --- a/internal/web/service/inbound_node.go +++ b/internal/web/service/inbound_node.go @@ -390,7 +390,9 @@ func (s *InboundService) SnapshotHasUnadoptedInbounds(nodeID int, snap *runtime. // config push just landed, whose snapshot may still predate it (#6228). func (s *InboundService) SetRemoteTraffic(nodeID int, snap *runtime.TrafficSnapshot, dirty, justPushed bool) (bool, error) { var structuralChange bool - err := submitTrafficWrite(func() error { + // Node merges take the background lane: a per-client mutation the operator + // is waiting on must not queue behind the next batch of 5s snapshots. + err := submitTrafficWriteBackground(func() error { var inner error structuralChange, inner = s.setRemoteTrafficLocked(nodeID, snap, dirty, justPushed) return inner diff --git a/internal/web/service/inbound_traffic.go b/internal/web/service/inbound_traffic.go index 3669ca3d4..7de298b98 100644 --- a/internal/web/service/inbound_traffic.go +++ b/internal/web/service/inbound_traffic.go @@ -28,7 +28,9 @@ func (s *InboundService) AddTraffic(inboundTraffics []*xray.Traffic, clientTraff var disabledNodeIDs []int var remotePlans []trafficInboundUpdatePlan var renewed []string - err = submitTrafficWrite(func() error { + // The periodic poll takes the background lane: an operator's save or reset + // queued while a poll runs must not sit behind the next merge batch. + err = submitTrafficWriteBackground(func() error { var inner error needRestart, clientsDisabled, disabledNodeIDs, remotePlans, renewed, inner = s.addTrafficLocked(inboundTraffics, clientTraffics) return inner @@ -720,55 +722,61 @@ func (s *InboundService) resetClientTrafficLocked(id int, clientEmail string) (b needRestart := false var reenablePlan *trafficLocalApplyPlan var reenableNodeID *int + // This inbound's stored entry read off and this reset flips it back on, + // in the same transaction as the zeroing. + flipStoredEntry := false traffic, err := s.GetClientTrafficByEmail(clientEmail) if err != nil { return false, nil, err } - // A reset lifts the quota gate only: a client switched off by hand stays off. keepOff, err := handDisabledEmails(database.GetDB(), []string{clientEmail}) if err != nil { return false, nil, err } handDisabled := len(keepOff) > 0 - - if !traffic.Enable && !handDisabled { - inbound, err := s.GetInbound(id) - if err != nil { - return false, nil, err + inbound, err := s.GetInbound(id) + if err != nil { + return false, nil, err + } + clients, err := s.GetClients(inbound) + if err != nil { + return false, nil, err + } + // Check each stored entry: the first reset already enables the shared + // row. A row-only check would leave sibling inbounds disabled. + entry := (*model.Client)(nil) + for i := range clients { + if clients[i].Email == clientEmail { + entry = &clients[i] + break } - clients, err := s.GetClients(inbound) - if err != nil { - return false, nil, err - } - for _, client := range clients { - if client.Email == clientEmail && client.Enable { - cipher := "" - if string(inbound.Protocol) == "shadowsocks" { - var oldSettings map[string]any - err = json.Unmarshal([]byte(inbound.Settings), &oldSettings) - if err != nil { - return false, nil, err - } - cipher, _ = oldSettings["method"].(string) - } - clientMap := map[string]any{ - "email": client.Email, - "id": client.ID, - "auth": client.Auth, - "security": client.Security, - "flow": client.Flow, - "password": client.Password, - "cipher": cipher, - "reverse": client.Reverse, - } - if inbound.NodeID != nil { - reenableNodeID = inbound.NodeID - } else { - reenablePlan = &trafficLocalApplyPlan{action: trafficAddUser, inbound: *inbound, client: clientMap} - } - break + } + if entry != nil && !handDisabled && (!entry.Enable || !traffic.Enable) { + flipStoredEntry = true + cipher := "" + if string(inbound.Protocol) == "shadowsocks" { + var oldSettings map[string]any + err = json.Unmarshal([]byte(inbound.Settings), &oldSettings) + if err != nil { + return false, nil, err } + cipher, _ = oldSettings["method"].(string) + } + clientMap := map[string]any{ + "email": entry.Email, + "id": entry.ID, + "auth": entry.Auth, + "security": entry.Security, + "flow": entry.Flow, + "password": entry.Password, + "cipher": cipher, + "reverse": entry.Reverse, + } + if inbound.NodeID != nil { + reenableNodeID = inbound.NodeID + } else { + reenablePlan = &trafficLocalApplyPlan{action: trafficAddUser, inbound: *inbound, client: clientMap} } } @@ -778,10 +786,6 @@ func (s *InboundService) resetClientTrafficLocked(id int, clientEmail string) (b db := database.GetDB() now := time.Now().UnixMilli() - inbound, err := s.GetInbound(id) - if err != nil { - return false, nil, err - } if err := db.Transaction(func(tx *gorm.DB) error { if err := adjustGroupBaselinesForRemovedTraffic(tx, []string{clientEmail}); err != nil { return err @@ -803,16 +807,32 @@ func (s *InboundService) resetClientTrafficLocked(id int, clientEmail string) (b Update("last_traffic_reset_time", now).Error; err != nil { return err } + if flipStoredEntry { + // Restore quota-disabled entries with the counters; an operator + // disable has already been excluded. + if err := s.markClientEnabledInSettings(tx, id, clientEmail); err != nil { + return err + } + if err := tx.Model(&model.ClientRecord{}).Where("email = ?", clientEmail). + Updates(map[string]any{"enable": true, "updated_at": now}).Error; err != nil { + return err + } + } if reenableNodeID != nil { return (&NodeService{}).MarkNodeDirtyTx(tx, *reenableNodeID) } - if inbound != nil && inbound.NodeID != nil { + if inbound.NodeID != nil { return (&NodeService{}).MarkNodeDirtyTx(tx, *inbound.NodeID) } return nil }); err != nil { return false, nil, err } + if flipStoredEntry { + // The entry left the subscription in its disabled state; a just + // re-enabled client must not keep missing from it. + InvalidateSubData() + } if reenablePlan != nil { rt, err := s.runtimeFor(&reenablePlan.inbound) @@ -826,10 +846,7 @@ func (s *InboundService) resetClientTrafficLocked(id int, clientEmail string) (b } } - if inbound != nil { - return needRestart, inbound.NodeID, nil - } - return needRestart, nil, nil + return needRestart, inbound.NodeID, nil } // ResetAllTraffics zeroes every inbound's counters and brings back the inbounds a traffic diff --git a/internal/web/service/inbound_traffic_global.go b/internal/web/service/inbound_traffic_global.go index 43eea2022..55fd08ffd 100644 --- a/internal/web/service/inbound_traffic_global.go +++ b/internal/web/service/inbound_traffic_global.go @@ -37,7 +37,7 @@ func (s *InboundService) AcceptGlobalTraffic(masterGuid string, traffics []*xray return nil } - return submitTrafficWrite(func() error { + return submitTrafficWriteBackground(func() error { db := database.GetDB() known := make([]string, 0, len(emails)) for _, batch := range chunkStrings(emails, sqlInChunk) { diff --git a/internal/web/service/traffic_writer.go b/internal/web/service/traffic_writer.go index e8c43d7a4..b65c9eea5 100644 --- a/internal/web/service/traffic_writer.go +++ b/internal/web/service/traffic_writer.go @@ -26,11 +26,12 @@ type trafficWriteRequest struct { type serializedTxContextKey struct{} var ( - twMu sync.Mutex - twQueue chan *trafficWriteRequest - twCtx context.Context - twCancel context.CancelFunc - twDone chan struct{} + twMu sync.Mutex + twAdminQueue chan *trafficWriteRequest + twBgQueue chan *trafficWriteRequest + twCtx context.Context + twCancel context.CancelFunc + twDone chan struct{} ) // StartTrafficWriter spins up the serial writer goroutine. Safe to call again @@ -53,16 +54,18 @@ func StartTrafficWriter() { } } - queue := make(chan *trafficWriteRequest, trafficWriterQueueSize) + adminQueue := make(chan *trafficWriteRequest, trafficWriterQueueSize) + bgQueue := make(chan *trafficWriteRequest, trafficWriterQueueSize) ctx, cancel := context.WithCancel(context.Background()) done := make(chan struct{}) - twQueue = queue + twAdminQueue = adminQueue + twBgQueue = bgQueue twCtx = ctx twCancel = cancel twDone = done - go runTrafficWriter(ctx, queue, done) + go runTrafficWriter(ctx, adminQueue, bgQueue, done) } // StopTrafficWriter cancels the writer context and waits for the goroutine to @@ -89,31 +92,68 @@ func StopTrafficWriter() { } func clearTrafficWriterState() { - twQueue = nil + twAdminQueue = nil + twBgQueue = nil twCtx = nil twCancel = nil twDone = nil } -func runTrafficWriter(ctx context.Context, queue chan *trafficWriteRequest, done chan struct{}) { +// trafficWriterQueueLen is the test's view of both lanes' pending requests. +func trafficWriterQueueLen() int { + twMu.Lock() + defer twMu.Unlock() + return len(twAdminQueue) + len(twBgQueue) +} + +// One goroutine serves both lanes (no two writes side by side — the Postgres +// deadlock guard c5d31de4e; the admin lane is preferred over the 5s poll and node merges, #6035). +func runTrafficWriter(ctx context.Context, adminQueue, bgQueue chan *trafficWriteRequest, done chan struct{}) { defer close(done) for { select { - case req := <-queue: + case req := <-adminQueue: req.done <- safeApply(req.apply) case <-ctx.Done(): - for { - select { - case req := <-queue: - req.done <- safeApply(req.apply) - default: - return - } + drainTrafficWriter(adminQueue, bgQueue) + return + default: + select { + case req := <-adminQueue: + req.done <- safeApply(req.apply) + case req := <-bgQueue: + req.done <- safeApply(req.apply) + case <-ctx.Done(): + drainTrafficWriter(adminQueue, bgQueue) + return } } } } +// drainTrafficWriter finishes every request already queued on shutdown, as the +// single-queue writer used to: both lanes drain until empty, so no write is dropped. +func drainTrafficWriter(adminQueue, bgQueue chan *trafficWriteRequest) { + for { + drained := false + select { + case req := <-adminQueue: + req.done <- safeApply(req.apply) + drained = true + default: + } + select { + case req := <-bgQueue: + req.done <- safeApply(req.apply) + drained = true + default: + } + if !drained { + return + } + } +} + // runSerializedTx runs fn inside one DB transaction on the shared serial // traffic-writer goroutine, so it can never execute concurrently with the // @every 5s traffic poll (AddTraffic). Both touch the hot client_traffics and @@ -167,13 +207,28 @@ func safeApply(fn func() error) (err error) { return fn() } +// submitTrafficWrite runs fn on the serial writer's admin lane, ahead of any +// queued periodic work (the traffic poll, the node sync merges). func submitTrafficWrite(fn func() error) error { + return submitTrafficWriteLane(fn, false) +} + +// submitTrafficWriteBackground runs fn on the periodic lane (the 5s traffic poll +// and node snapshot merges), served only when no admin-lane write is queued. +func submitTrafficWriteBackground(fn func() error) error { + return submitTrafficWriteLane(fn, true) +} + +func submitTrafficWriteLane(fn func() error, background bool) error { req := &trafficWriteRequest{apply: fn, done: make(chan error, 1)} twMu.Lock() - queue := twQueue + queue := twAdminQueue ctx := twCtx done := twDone + if background { + queue = twBgQueue + } if queue == nil || ctx == nil || done == nil { twMu.Unlock() return safeApply(fn) diff --git a/internal/web/service/traffic_writer_test.go b/internal/web/service/traffic_writer_test.go index 42520fbbe..8623cf4b1 100644 --- a/internal/web/service/traffic_writer_test.go +++ b/internal/web/service/traffic_writer_test.go @@ -161,13 +161,7 @@ func waitTrafficWriterQueued(t *testing.T) { deadline := time.Now().Add(time.Second) for time.Now().Before(deadline) { - twMu.Lock() - queued := 0 - if twQueue != nil { - queued = len(twQueue) - } - twMu.Unlock() - if queued > 0 { + if trafficWriterQueueLen() > 0 { return } time.Sleep(10 * time.Millisecond) @@ -217,10 +211,7 @@ func parkSerializedWriter(t *testing.T) (release func()) { func waitForQueuedWrite() bool { deadline := time.Now().Add(time.Second) for time.Now().Before(deadline) { - twMu.Lock() - queued := len(twQueue) > 0 - twMu.Unlock() - if queued { + if trafficWriterQueueLen() > 0 { return true } time.Sleep(10 * time.Millisecond) @@ -228,6 +219,99 @@ func waitForQueuedWrite() bool { return false } +// A write the operator is waiting on must not queue behind the periodic work +// that keeps arriving every 5s (#6035): with both lanes filled, admin runs first. +func TestTrafficWriterAdminLanePreemptsBackgroundQueue(t *testing.T) { + resetTrafficWriterForTest(t) + + StartTrafficWriter() + parkStarted := make(chan struct{}) + releasePark := make(chan struct{}) + go func() { + _ = submitTrafficWrite(func() error { + close(parkStarted) + <-releasePark + return nil + }) + }() + waitTrafficWriterSignal(t, parkStarted, "parking write did not start") + + order := make(chan string, 2) + bgDone := make(chan error, 1) + go func() { + bgDone <- submitTrafficWriteBackground(func() error { + order <- "background" + return nil + }) + }() + adminDone := make(chan error, 1) + go func() { + adminDone <- submitTrafficWrite(func() error { + order <- "admin" + return nil + }) + }() + deadline := time.Now().Add(time.Second) + for trafficWriterQueueLen() < 2 && time.Now().Before(deadline) { + time.Sleep(5 * time.Millisecond) + } + if trafficWriterQueueLen() < 2 { + t.Fatal("both writes did not reach the queues") + } + + close(releasePark) + if got := <-order; got != "admin" { + t.Fatalf("first write served after the parked one = %q, want admin: the lane the operator waits on", got) + } + if got := <-order; got != "background" { + t.Fatalf("second write served = %q, want background", got) + } + if err := <-adminDone; err != nil { + t.Fatalf("admin write: %v", err) + } + if err := <-bgDone; err != nil { + t.Fatalf("background write: %v", err) + } +} + +// A stop must drain every queued write on both lanes, not one per lane: a +// submitter blocked behind two queued writes must not read "stopped". +func TestTrafficWriterStopDrainsEveryQueuedWrite(t *testing.T) { + resetTrafficWriterForTest(t) + + StartTrafficWriter() + parkStarted := make(chan struct{}) + releasePark := make(chan struct{}) + go func() { + _ = submitTrafficWrite(func() error { + close(parkStarted) + <-releasePark + return nil + }) + }() + waitTrafficWriterSignal(t, parkStarted, "parking write did not start") + + results := make(chan error, 3) + go func() { results <- submitTrafficWrite(func() error { return nil }) }() + go func() { results <- submitTrafficWrite(func() error { return nil }) }() + go func() { results <- submitTrafficWriteBackground(func() error { return nil }) }() + deadline := time.Now().Add(time.Second) + for trafficWriterQueueLen() < 3 && time.Now().Before(deadline) { + time.Sleep(5 * time.Millisecond) + } + if trafficWriterQueueLen() < 3 { + t.Fatal("queued writes did not all reach the lanes") + } + + close(releasePark) + StopTrafficWriter() + for range 3 { + if err := waitTrafficWriterErr(t, results); err != nil { + t.Fatalf("a queued write was dropped on stop: %v", err) + } + } +} + func waitWriterResult(t *testing.T, done <-chan error) error { t.Helper() select { diff --git a/internal/web/service/tuic_traffic_receipt.go b/internal/web/service/tuic_traffic_receipt.go index 117df3785..35b925cac 100644 --- a/internal/web/service/tuic_traffic_receipt.go +++ b/internal/web/service/tuic_traffic_receipt.go @@ -17,7 +17,8 @@ func (s *InboundService) AddTuicTrafficBatch(id string, traffic []*xray.ClientTr if id == "" { return errors.New("TUIC traffic batch has no ID") } - return submitTrafficWrite(func() error { + // The receipt journal is periodic accounting, not an operator action. + return submitTrafficWriteBackground(func() error { return database.GetDB().Transaction(func(tx *gorm.DB) error { result := tx.Clauses(clause.OnConflict{DoNothing: true}).Create(&model.TuicTrafficReceipt{ID: id}) if result.Error != nil {