From eae4725d99e7274f1b78b78fc1b56179e03ba2ce Mon Sep 17 00:00:00 2001 From: 25f2002654 Date: Mon, 2 Nov 2026 20:20:25 +0530 Subject: [PATCH] fix(traffic): restore quota-disabled inbounds when their traffic is reset (#6962) An inbound switched off by its own traffic cap came back on the periodic reset, but the admin per-inbound reset and Reset All Traffic only zeroed the counters. After that nothing could tell it from a hand-disabled inbound, so it stayed off with every client on it. Both admin resets now re-enable the inbounds their cap switched off, using the predicate the periodic job used, now in the service next to disableInvalidInbounds; expired and hand-disabled inbounds stay off. --- internal/web/controller/inbound.go | 10 +- .../web/job/periodic_traffic_reset_job.go | 22 +--- .../web/service/api_scale_postgres_test.go | 4 +- internal/web/service/inbound_disable.go | 24 ++++ internal/web/service/inbound_traffic.go | 53 +++++++-- .../service/inbound_traffic_restore_test.go | 103 ++++++++++++++++++ .../web/service/mtproto_quota_reset_test.go | 2 +- .../web/service/node_admin_fanout_test.go | 2 +- internal/web/service/reset_writer_test.go | 2 +- 9 files changed, 184 insertions(+), 38 deletions(-) create mode 100644 internal/web/service/inbound_traffic_restore_test.go diff --git a/internal/web/controller/inbound.go b/internal/web/controller/inbound.go index da438fbab..c1387f7d9 100644 --- a/internal/web/controller/inbound.go +++ b/internal/web/controller/inbound.go @@ -340,11 +340,12 @@ func (a *InboundController) resetInboundTraffic(c *gin.Context) { return } - err = a.inboundService.ResetInboundTraffic(id) + needRestart, err := a.inboundService.ResetInboundTraffic(id) if err != nil { jsonMsg(c, I18nWeb(c, "somethingWentWrong"), err) return - } else { + } + if needRestart { a.xrayService.SetToNeedRestart() } jsonMsg(c, I18nWeb(c, "pages.inbounds.toasts.resetInboundTrafficSuccess"), nil) @@ -386,11 +387,12 @@ func (a *InboundController) delAllInboundClients(c *gin.Context) { // resetAllTraffics resets all traffic counters across all inbounds. func (a *InboundController) resetAllTraffics(c *gin.Context) { - err := a.inboundService.ResetAllTraffics() + needRestart, err := a.inboundService.ResetAllTraffics() if err != nil { jsonMsg(c, I18nWeb(c, "somethingWentWrong"), err) return - } else { + } + if needRestart { a.xrayService.SetToNeedRestart() } jsonMsg(c, I18nWeb(c, "pages.inbounds.toasts.resetAllTrafficSuccess"), nil) diff --git a/internal/web/job/periodic_traffic_reset_job.go b/internal/web/job/periodic_traffic_reset_job.go index bc43b7d90..48f467cc9 100644 --- a/internal/web/job/periodic_traffic_reset_job.go +++ b/internal/web/job/periodic_traffic_reset_job.go @@ -5,7 +5,6 @@ import ( "sync/atomic" "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/util/common" "github.com/mhsanaei/3x-ui/v3/internal/web/service" @@ -43,13 +42,6 @@ func monthlyResetDue(resetDay int, now time.Time) bool { return now.Day() == min(resetDay, lastDay) } -// inboundQuotaDisabled mirrors disableInvalidInbounds: a local inbound its own -// traffic cap switched off, not one that expired or was switched off by hand. -func inboundQuotaDisabled(ib *model.Inbound, nowMs int64) bool { - return !ib.Enable && ib.NodeID == nil && ib.Total > 0 && ib.Up+ib.Down >= ib.Total && - (ib.ExpiryTime <= 0 || ib.ExpiryTime > nowMs) -} - func forEachResetBounded(n int, reset func(i int)) { sem := make(chan struct{}, periodicResetConcurrency) var wg sync.WaitGroup @@ -96,21 +88,13 @@ func (j *PeriodicTrafficResetJob) resetInboundsOnSchedule() { var resetCount atomic.Int32 var needRestart atomic.Bool - nowMs := time.Now().UnixMilli() forEachResetBounded(len(inbounds), func(i int) { inbound := inbounds[i] - quotaOff := inboundQuotaDisabled(inbound, nowMs) - resetInboundErr := j.inboundService.ResetInboundTraffic(inbound.Id) + nr, resetInboundErr := j.inboundService.ResetInboundTraffic(inbound.Id) if resetInboundErr != nil { logger.Warning("Failed to reset traffic for inbound", inbound.Id, ":", resetInboundErr) - } else if quotaOff { - nr, err := j.inboundService.SetInboundEnable(inbound.Id, true) - if err != nil { - logger.Warning("Failed to re-enable inbound", inbound.Id, "after traffic reset:", err) - } - if nr { - needRestart.Store(true) - } + } else if nr { + needRestart.Store(true) } nr, resetClientErr := j.clientService.ResetAllClientTraffics(&j.inboundService, inbound.Id) diff --git a/internal/web/service/api_scale_postgres_test.go b/internal/web/service/api_scale_postgres_test.go index 3448ba443..2b65020b9 100644 --- a/internal/web/service/api_scale_postgres_test.go +++ b/internal/web/service/api_scale_postgres_test.go @@ -105,8 +105,8 @@ func TestAllAPIsPostgresScale(t *testing.T) { run("RenameGroup", func() error { _, err := svc.RenameGroup("g1", "g2"); return err }) run("DeleteGroup", func() error { _, err := svc.DeleteGroup("g2"); return err }) - run("ResetInboundTraffic", func() error { return inboundSvc.ResetInboundTraffic(ib.Id) }) - run("Inbound.ResetAllTraffics", func() error { return inboundSvc.ResetAllTraffics() }) + run("ResetInboundTraffic", func() error { _, err := inboundSvc.ResetInboundTraffic(ib.Id); return err }) + run("Inbound.ResetAllTraffics", func() error { _, err := inboundSvc.ResetAllTraffics(); return err }) run("Client.ResetAllTraffics", func() error { _, err := svc.ResetAllTraffics(); return err }) run("BulkResetTraffic(M)", func() error { _, err := svc.BulkResetTraffic(inboundSvc, emailsM); return err }) diff --git a/internal/web/service/inbound_disable.go b/internal/web/service/inbound_disable.go index 46cdbba51..597b8853b 100644 --- a/internal/web/service/inbound_disable.go +++ b/internal/web/service/inbound_disable.go @@ -35,6 +35,30 @@ func (s *InboundService) disableInvalidInbounds(tx *gorm.DB, mutationBatch *traf return false, count, err } +// quotaDisabledInboundCond matches local inbounds their own cap switched off (args: enable, nowMs); +// expired or hand-disabled ones never match, so a reset cannot bring them back. +const quotaDisabledInboundCond = `enable = ? AND node_id IS NULL AND total > 0 AND up + down >= total AND (expiry_time <= 0 OR expiry_time > ?)` + +// quotaDisabledInboundIDs lists the inbounds a traffic cap switched off. +func quotaDisabledInboundIDs(db *gorm.DB, nowMs int64) ([]int, error) { + var ids []int + err := db.Model(model.Inbound{}). + Where("user_id > ?", 0). + Where(quotaDisabledInboundCond, false, nowMs). + Pluck("id", &ids).Error + return ids, err +} + +// quotaDisabledInbound reports whether a traffic cap switched this inbound off. +func quotaDisabledInbound(db *gorm.DB, id int, nowMs int64) (bool, error) { + var count int64 + err := db.Model(model.Inbound{}). + Where("id = ?", id). + Where(quotaDisabledInboundCond, false, nowMs). + Count(&count).Error + return count > 0, err +} + const globalTrafficFreshWindow = 24 * time.Hour func globalTrafficFreshSince(now time.Time) int64 { diff --git a/internal/web/service/inbound_traffic.go b/internal/web/service/inbound_traffic.go index 9fd889c4e..3669ca3d4 100644 --- a/internal/web/service/inbound_traffic.go +++ b/internal/web/service/inbound_traffic.go @@ -832,14 +832,33 @@ func (s *InboundService) resetClientTrafficLocked(id int, clientEmail string) (b return needRestart, nil, nil } -func (s *InboundService) ResetAllTraffics() error { +// ResetAllTraffics zeroes every inbound's counters and brings back the inbounds a traffic +// cap switched off; left off, every client on them would stay dead. +func (s *InboundService) ResetAllTraffics() (bool, error) { + var restore []int err := submitTrafficWrite(func() error { + db := database.GetDB() + var err error + restore, err = quotaDisabledInboundIDs(db, time.Now().UnixMilli()) + if err != nil { + return err + } return s.resetAllTrafficsLocked() }) - if err == nil { - s.propagateResetAllTrafficsToNodes() + if err != nil { + return false, err } - return err + needRestart := false + for _, id := range restore { + nr, err := s.SetInboundEnable(id, true) + if err != nil { + logger.Warning("ResetAllTraffics: re-enable quota-disabled inbound", id, "failed:", err) + continue + } + needRestart = needRestart || nr + } + s.propagateResetAllTrafficsToNodes() + return needRestart, nil } func (s *InboundService) resetAllTrafficsLocked() error { @@ -878,23 +897,37 @@ func (s *InboundService) propagateResetAllTrafficsToNodes() { }) } -func (s *InboundService) ResetInboundTraffic(id int) error { +// ResetInboundTraffic zeroes one inbound's counters and brings it back when +// its own traffic cap was what switched it off. +func (s *InboundService) ResetInboundTraffic(id int) (bool, error) { var inbound *model.Inbound + var quotaOff bool if err := submitTrafficWrite(func() error { db := database.GetDB() - if err := db.Model(model.Inbound{}). - Where("id = ?", id). + var err error + quotaOff, err = quotaDisabledInbound(db, id, time.Now().UnixMilli()) + if err != nil { + return err + } + if err := db.Model(model.Inbound{}).Where("id = ?", id). 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 + return false, err + } + needRestart := false + if quotaOff { + nr, err := s.SetInboundEnable(id, true) + if err != nil { + logger.Warning("ResetInboundTraffic: re-enable quota-disabled inbound", id, "failed:", err) + } + needRestart = needRestart || nr } if inbound != nil && inbound.NodeID != nil { if rt, rterr := s.runtimeFor(inbound); rterr == nil { @@ -905,7 +938,7 @@ func (s *InboundService) ResetInboundTraffic(id int) error { logger.Warning("ResetInboundTraffic: runtime lookup failed:", rterr) } } - return nil + return needRestart, nil } func (s *InboundService) DelDepletedClients(id int) (err error) { diff --git a/internal/web/service/inbound_traffic_restore_test.go b/internal/web/service/inbound_traffic_restore_test.go new file mode 100644 index 000000000..c9701db36 --- /dev/null +++ b/internal/web/service/inbound_traffic_restore_test.go @@ -0,0 +1,103 @@ +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/web/runtime" +) + +// setupQuotaRestoreTest wires a temp DB and a local runtime that accepts +// inbound changes, so a reset can be driven end to end. +func setupQuotaRestoreTest(t *testing.T) { + t.Helper() + initTrafficTestDB(t) + resetTrafficWriterForTest(t) + + mgr := runtime.NewManager(runtime.LocalDeps{APIPort: func() int { return 0 }}) + mgr.SetLocalRuntimeOverride(&fakeNodeRuntime{}) + runtime.SetManager(mgr) + t.Cleanup(func() { runtime.SetManager(nil) }) +} + +// seedQuotaInbound creates an inbound in the state a traffic cap leaves it in: +// switched off, with its counters past the cap. +func seedQuotaInbound(t *testing.T, tag string, port int, total, up int64, expiryTime int64) *model.Inbound { + t.Helper() + ib := &model.Inbound{ + UserId: 1, + Tag: tag, + Enable: false, + Listen: "0.0.0.0", + Port: port, + Protocol: model.VLESS, + StreamSettings: `{"network":"tcp"}`, + Settings: `{}`, + Total: total, + Up: up, + ExpiryTime: expiryTime, + } + if err := database.GetDB().Create(ib).Error; err != nil { + t.Fatalf("seed inbound %s: %v", tag, err) + } + return ib +} + +// A cap switches an inbound off and out of the core; zeroing the counters lifts +// the cap, so the inbound must come back or every client on it stays down. +func TestResetAllTrafficsRestoresQuotaDisabledInbound(t *testing.T) { + setupQuotaRestoreTest(t) + const gb = int64(1024 * 1024 * 1024) + seedQuotaInbound(t, "capped", 45501, gb, gb, 0) + + if _, err := (&InboundService{}).ResetAllTraffics(); err != nil { + t.Fatalf("ResetAllTraffics: %v", err) + } + + after := loadInboundByTag(t, "capped") + if !after.Enable { + t.Fatal("an inbound whose traffic cap was lifted must be enabled again") + } + if after.Up != 0 || after.Down != 0 { + t.Fatalf("counters = up %d / down %d, want both zero", after.Up, after.Down) + } +} + +func TestResetInboundTrafficRestoresQuotaDisabledInbound(t *testing.T) { + setupQuotaRestoreTest(t) + const gb = int64(1024 * 1024 * 1024) + ib := seedQuotaInbound(t, "capped", 45502, gb, gb, 0) + + if _, err := (&InboundService{}).ResetInboundTraffic(ib.Id); err != nil { + t.Fatalf("ResetInboundTraffic: %v", err) + } + + after := loadInboundByTag(t, "capped") + if !after.Enable { + t.Fatal("an inbound whose traffic cap was lifted must be enabled again") + } + if after.Up != 0 || after.Down != 0 { + t.Fatalf("counters = up %d / down %d, want both zero", after.Up, after.Down) + } +} + +// Only a cap may bring an inbound back. One an operator switched off by hand, +// and one that expired, stay off — the reset has nothing to do with either. +func TestResetAllTrafficsLeavesHandDisabledAndExpiredInboundsOff(t *testing.T) { + setupQuotaRestoreTest(t) + const gb = int64(1024 * 1024 * 1024) + seedQuotaInbound(t, "hand-disabled", 45503, 0, 0, 0) + seedQuotaInbound(t, "expired", 45504, gb, gb, time.Now().Add(-time.Hour).UnixMilli()) + + if _, err := (&InboundService{}).ResetAllTraffics(); err != nil { + t.Fatalf("ResetAllTraffics: %v", err) + } + + for _, tag := range []string{"hand-disabled", "expired"} { + if after := loadInboundByTag(t, tag); after.Enable { + t.Fatalf("inbound %q must stay disabled", tag) + } + } +} diff --git a/internal/web/service/mtproto_quota_reset_test.go b/internal/web/service/mtproto_quota_reset_test.go index 5b87109bf..34f6c40b2 100644 --- a/internal/web/service/mtproto_quota_reset_test.go +++ b/internal/web/service/mtproto_quota_reset_test.go @@ -112,7 +112,7 @@ func TestPanelResetsZeroSidecarQuota(t *testing.T) { // sidecar's quota counters must stay too or clients get their quota again free. func TestInboundResetAllKeepsSidecarQuota(t *testing.T) { _, logPath := startQuotaSidecar(t, 46204, model.Client{Enable: true}) - if err := (&InboundService{}).ResetAllTraffics(); err != nil { + if _, err := (&InboundService{}).ResetAllTraffics(); err != nil { t.Fatalf("ResetAllTraffics: %v", err) } if got := quotaResets(t, logPath); len(got) != 0 { diff --git a/internal/web/service/node_admin_fanout_test.go b/internal/web/service/node_admin_fanout_test.go index 2930d94b5..561eaed30 100644 --- a/internal/web/service/node_admin_fanout_test.go +++ b/internal/web/service/node_admin_fanout_test.go @@ -85,7 +85,7 @@ func TestResetAllTrafficsReachesNodesConcurrently(t *testing.T) { done := make(chan struct{}) go func() { defer close(done) - _ = (&InboundService{}).ResetAllTraffics() + _, _ = (&InboundService{}).ResetAllTraffics() }() t.Cleanup(func() { gate.open(); <-done }) gate.waitAll(t, 3, "ResetAllTraffics") diff --git a/internal/web/service/reset_writer_test.go b/internal/web/service/reset_writer_test.go index 0ffa13192..c57152020 100644 --- a/internal/web/service/reset_writer_test.go +++ b/internal/web/service/reset_writer_test.go @@ -38,7 +38,7 @@ func TestResetAllTrafficsDoesNotBlockWriterOnNodeCall(t *testing.T) { mgr.SetRuntimeOverride(node.Id, fake) done := make(chan error, 1) - go func() { done <- (&InboundService{}).ResetAllTraffics() }() + go func() { _, err := (&InboundService{}).ResetAllTraffics(); done <- err }() select { case <-fake.reached: