优化并发槽位清理

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()) {

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
 }

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
 }

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

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)

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() {

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
 }

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:
+		}
+	}
+}

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
 }

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
 }