fix(clients): restore depleted clients on renewal and prioritize operator writes (#6987)

Restore usable clients when renewed, avoid duplicate reset updates, and prioritize operator writes in the serialized traffic writer. Preserve manual disables and keep every inbound consistent after traffic resets.
This commit is contained in:
Mahdi Fallahnejhad
2026-11-05 20:37:52 +03:30
committed by GitHub
parent e4935be861
commit dbb6212988
13 changed files with 684 additions and 92 deletions
+15
View File
@@ -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 := ""
+28 -1
View File
@@ -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
}
})
}
@@ -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
}
+15 -10
View File
@@ -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
@@ -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)
@@ -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)
}
}
+42
View File
@@ -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) {
+3 -1
View File
@@ -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
+64 -47
View File
@@ -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
@@ -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) {
+74 -19
View File
@@ -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)
+95 -11
View File
@@ -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 {
+2 -1
View File
@@ -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 {