diff --git a/backend/internal/handler/gateway_handler_warmup_intercept_unit_test.go b/backend/internal/handler/gateway_handler_warmup_intercept_unit_test.go index 09b20722c9..75e3c1aa8c 100644 --- a/backend/internal/handler/gateway_handler_warmup_intercept_unit_test.go +++ b/backend/internal/handler/gateway_handler_warmup_intercept_unit_test.go @@ -137,6 +137,7 @@ func (f *fakeConcurrencyCache) GetAccountConcurrencyBatch(_ context.Context, acc return result, nil } func (f *fakeConcurrencyCache) CleanupExpiredAccountSlots(context.Context, int64) error { return nil } +func (f *fakeConcurrencyCache) CleanupExpiredAccountSlotKeys(context.Context) error { return nil } func (f *fakeConcurrencyCache) CleanupStaleProcessSlots(context.Context, string) error { return nil } func newTestGatewayHandler(t *testing.T, group *service.Group, accounts []*service.Account) (*GatewayHandler, func()) { diff --git a/backend/internal/handler/gateway_helper_fastpath_test.go b/backend/internal/handler/gateway_helper_fastpath_test.go index c7c0fb6c9e..fecb9b071d 100644 --- a/backend/internal/handler/gateway_helper_fastpath_test.go +++ b/backend/internal/handler/gateway_helper_fastpath_test.go @@ -89,6 +89,10 @@ func (m *concurrencyCacheMock) CleanupExpiredAccountSlots(ctx context.Context, a return nil } +func (m *concurrencyCacheMock) CleanupExpiredAccountSlotKeys(ctx context.Context) error { + return nil +} + func (m *concurrencyCacheMock) CleanupStaleProcessSlots(ctx context.Context, activeRequestPrefix string) error { return nil } diff --git a/backend/internal/handler/gateway_helper_hotpath_test.go b/backend/internal/handler/gateway_helper_hotpath_test.go index 65dc849683..fb17481f1c 100644 --- a/backend/internal/handler/gateway_helper_hotpath_test.go +++ b/backend/internal/handler/gateway_helper_hotpath_test.go @@ -140,6 +140,10 @@ func (s *helperConcurrencyCacheStub) CleanupExpiredAccountSlots(ctx context.Cont return nil } +func (s *helperConcurrencyCacheStub) CleanupExpiredAccountSlotKeys(ctx context.Context) error { + return nil +} + func (s *helperConcurrencyCacheStub) CleanupStaleProcessSlots(ctx context.Context, activeRequestPrefix string) error { return nil } diff --git a/backend/internal/repository/concurrency_cache.go b/backend/internal/repository/concurrency_cache.go index 5e6f10062f..b530c30b90 100644 --- a/backend/internal/repository/concurrency_cache.go +++ b/backend/internal/repository/concurrency_cache.go @@ -174,6 +174,29 @@ var ( return 1 `) + // cleanupExpiredSlotKeysScript 批量清理实际存在的账号槽位键,避免后台任务从数据库加载全量账号。 + // KEYS = 有序集合键列表,ARGV[1] = TTL(秒)。 + cleanupExpiredSlotKeysScript = redis.NewScript(` + -- Redis 3.2-4.x compat: opt into effects replication so redis.call('TIME') + -- replicates correctly. No-op on Redis 5.0+ (effects replication is default). + redis.replicate_commands() + local ttl = tonumber(ARGV[1]) + local timeResult = redis.call('TIME') + local now = tonumber(timeResult[1]) + local expireBefore = now - ttl + local removed = 0 + for i = 1, #KEYS do + local key = KEYS[i] + removed = removed + redis.call('ZREMRANGEBYSCORE', key, '-inf', expireBefore) + if redis.call('ZCARD', key) == 0 then + redis.call('DEL', key) + else + redis.call('EXPIRE', key, ttl) + end + end + return removed + `) + // startupCleanupScript 清理非当前进程前缀的槽位成员。 // KEYS 是有序集合键列表,ARGV[1] 是当前进程前缀,ARGV[2] 是槽位 TTL。 // 遍历每个 KEYS[i],移除前缀不匹配的成员,清空后删 key,否则刷新 EXPIRE。 @@ -503,6 +526,10 @@ func (c *concurrencyCache) CleanupExpiredAccountSlots(ctx context.Context, accou return err } +func (c *concurrencyCache) CleanupExpiredAccountSlotKeys(ctx context.Context) error { + return c.cleanupExpiredSlotKeysByPattern(ctx, accountSlotKeyPrefix+"*") +} + func (c *concurrencyCache) CleanupStaleProcessSlots(ctx context.Context, activeRequestPrefix string) error { if activeRequestPrefix == "" { return nil @@ -527,6 +554,29 @@ func (c *concurrencyCache) CleanupStaleProcessSlots(ctx context.Context, activeR return nil } +// cleanupExpiredSlotKeysByPattern 扫描实际存在的账号槽位键并批量清理过期成员。 +func (c *concurrencyCache) cleanupExpiredSlotKeysByPattern(ctx context.Context, pattern string) error { + const scanCount = 200 + var cursor uint64 + for { + keys, nextCursor, err := c.rdb.Scan(ctx, cursor, pattern, scanCount).Result() + if err != nil { + return fmt.Errorf("scan %s: %w", pattern, err) + } + if len(keys) > 0 { + _, err := cleanupExpiredSlotKeysScript.Run(ctx, c.rdb, keys, c.slotTTLSeconds).Result() + if err != nil { + return fmt.Errorf("cleanup expired slots %s: %w", pattern, err) + } + } + cursor = nextCursor + if cursor == 0 { + break + } + } + return nil +} + // cleanupSlotsByPattern 扫描匹配 pattern 的有序集合键,批量调用 Lua 脚本清理非当前进程成员。 func (c *concurrencyCache) cleanupSlotsByPattern(ctx context.Context, pattern, activePrefix string) error { const scanCount = 200 diff --git a/backend/internal/repository/concurrency_cache_integration_test.go b/backend/internal/repository/concurrency_cache_integration_test.go index 5da94fc258..4aa172b252 100644 --- a/backend/internal/repository/concurrency_cache_integration_test.go +++ b/backend/internal/repository/concurrency_cache_integration_test.go @@ -435,6 +435,39 @@ func (s *ConcurrencyCacheSuite) TestCleanupExpiredAccountSlots_NoExpired() { require.Equal(s.T(), 2, cur) } +func (s *ConcurrencyCacheSuite) TestCleanupExpiredAccountSlotKeys() { + now := time.Now().Unix() + expiredTime := now - int64(testSlotTTL.Seconds()) - 10 + accountKeyWithFresh := fmt.Sprintf("%s%d", accountSlotKeyPrefix, 301) + accountKeyExpiredOnly := fmt.Sprintf("%s%d", accountSlotKeyPrefix, 302) + userKey := fmt.Sprintf("%s%d", userSlotKeyPrefix, 303) + + require.NoError(s.T(), s.rdb.ZAdd(s.ctx, accountKeyWithFresh, + redis.Z{Score: float64(expiredTime), Member: "expired"}, + redis.Z{Score: float64(now), Member: "fresh"}, + ).Err()) + require.NoError(s.T(), s.rdb.ZAdd(s.ctx, accountKeyExpiredOnly, + redis.Z{Score: float64(expiredTime), Member: "expired-only"}, + ).Err()) + require.NoError(s.T(), s.rdb.ZAdd(s.ctx, userKey, + redis.Z{Score: float64(expiredTime), Member: "user-expired"}, + ).Err()) + + require.NoError(s.T(), s.cache.CleanupExpiredAccountSlotKeys(s.ctx)) + + accountMembers, err := s.rdb.ZRange(s.ctx, accountKeyWithFresh, 0, -1).Result() + require.NoError(s.T(), err) + require.Equal(s.T(), []string{"fresh"}, accountMembers) + + exists, err := s.rdb.Exists(s.ctx, accountKeyExpiredOnly).Result() + require.NoError(s.T(), err) + require.EqualValues(s.T(), 0, exists) + + userMembers, err := s.rdb.ZRange(s.ctx, userKey, 0, -1).Result() + require.NoError(s.T(), err) + require.Equal(s.T(), []string{"user-expired"}, userMembers) +} + func (s *ConcurrencyCacheSuite) TestCleanupStaleProcessSlots_RemovesOldPrefixesAndWaitCounters() { accountID := int64(901) userID := int64(902) diff --git a/backend/internal/service/concurrency_service.go b/backend/internal/service/concurrency_service.go index 712fc1a749..50e0077b94 100644 --- a/backend/internal/service/concurrency_service.go +++ b/backend/internal/service/concurrency_service.go @@ -47,6 +47,7 @@ type ConcurrencyCache interface { // 清理过期槽位(后台任务) CleanupExpiredAccountSlots(ctx context.Context, accountID int64) error + CleanupExpiredAccountSlotKeys(ctx context.Context) error // 启动时清理旧进程遗留槽位与等待计数 CleanupStaleProcessSlots(ctx context.Context, activeRequestPrefix string) error @@ -473,27 +474,19 @@ func (s *ConcurrencyService) CleanupExpiredAccountSlots(ctx context.Context, acc } // StartSlotCleanupWorker starts a background cleanup worker for expired account slots. -func (s *ConcurrencyService) StartSlotCleanupWorker(accountRepo AccountRepository, interval time.Duration) { - if s == nil || s.cache == nil || accountRepo == nil || interval <= 0 { +func (s *ConcurrencyService) StartSlotCleanupWorker(_ AccountRepository, interval time.Duration) { + if s == nil || s.cache == nil || interval <= 0 { return } runCleanup := func() { - listCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - accounts, err := accountRepo.ListSchedulable(listCtx) + cleanupCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + err := s.cache.CleanupExpiredAccountSlotKeys(cleanupCtx) cancel() if err != nil { - logger.LegacyPrintf("service.concurrency", "Warning: list schedulable accounts failed: %v", err) + logger.LegacyPrintf("service.concurrency", "Warning: cleanup expired account slots failed: %v", err) return } - for _, account := range accounts { - accountCtx, accountCancel := context.WithTimeout(context.Background(), 2*time.Second) - err := s.cache.CleanupExpiredAccountSlots(accountCtx, account.ID) - accountCancel() - if err != nil { - logger.LegacyPrintf("service.concurrency", "Warning: cleanup expired slots failed for account %d: %v", account.ID, err) - } - } } go func() { diff --git a/backend/internal/service/concurrency_service_test.go b/backend/internal/service/concurrency_service_test.go index 7d5f501dc5..bacad0245e 100644 --- a/backend/internal/service/concurrency_service_test.go +++ b/backend/internal/service/concurrency_service_test.go @@ -95,6 +95,10 @@ func (c *stubConcurrencyCacheForTest) CleanupExpiredAccountSlots(_ context.Conte return c.cleanupErr } +func (c *stubConcurrencyCacheForTest) CleanupExpiredAccountSlotKeys(_ context.Context) error { + return c.cleanupErr +} + func (c *stubConcurrencyCacheForTest) CleanupStaleProcessSlots(_ context.Context, _ string) error { return c.cleanupErr } diff --git a/backend/internal/service/concurrency_slot_cleanup_test.go b/backend/internal/service/concurrency_slot_cleanup_test.go new file mode 100644 index 0000000000..be26bb391b --- /dev/null +++ b/backend/internal/service/concurrency_slot_cleanup_test.go @@ -0,0 +1,39 @@ +package service + +import ( + "context" + "sync/atomic" + "testing" + "time" +) + +type slotCleanupCache struct { + ConcurrencyCache + calls atomic.Int64 +} + +func (c *slotCleanupCache) CleanupExpiredAccountSlotKeys(context.Context) error { + c.calls.Add(1) + return nil +} + +func TestStartSlotCleanupWorker_UsesCacheWideCleanupWithoutAccountRepo(t *testing.T) { + cache := &slotCleanupCache{} + svc := NewConcurrencyService(cache) + + svc.StartSlotCleanupWorker(nil, time.Hour) + + deadline := time.After(time.Second) + ticker := time.NewTicker(10 * time.Millisecond) + defer ticker.Stop() + for { + if cache.calls.Load() > 0 { + return + } + select { + case <-deadline: + t.Fatal("cleanup worker did not call cache-wide account slot cleanup") + case <-ticker.C: + } + } +} diff --git a/backend/internal/service/gateway_multiplatform_test.go b/backend/internal/service/gateway_multiplatform_test.go index 291cea53b9..f843ba3e45 100644 --- a/backend/internal/service/gateway_multiplatform_test.go +++ b/backend/internal/service/gateway_multiplatform_test.go @@ -2097,6 +2097,10 @@ func (m *mockConcurrencyCache) CleanupExpiredAccountSlots(ctx context.Context, a return nil } +func (m *mockConcurrencyCache) CleanupExpiredAccountSlotKeys(ctx context.Context) error { + return nil +} + func (m *mockConcurrencyCache) CleanupStaleProcessSlots(ctx context.Context, activeRequestPrefix string) error { return nil } diff --git a/backend/internal/testutil/stubs.go b/backend/internal/testutil/stubs.go index bc572e1137..a0bd4bc669 100644 --- a/backend/internal/testutil/stubs.go +++ b/backend/internal/testutil/stubs.go @@ -76,6 +76,9 @@ func (c StubConcurrencyCache) GetAccountConcurrencyBatch(_ context.Context, acco func (c StubConcurrencyCache) CleanupExpiredAccountSlots(_ context.Context, _ int64) error { return nil } +func (c StubConcurrencyCache) CleanupExpiredAccountSlotKeys(_ context.Context) error { + return nil +} func (c StubConcurrencyCache) CleanupStaleProcessSlots(_ context.Context, _ string) error { return nil }