mirror of
https://gitee.com/wanwujie/sub2api
synced 2026-04-28 18:34:47 +08:00
feat: cleanup stale concurrency slots on startup
When the service restarts, concurrency slots from the old process remain in Redis, causing phantom occupancy. On startup, scan all concurrency sorted sets and remove members with non-current process prefix, then clear orphaned wait queue counters. Uses Go-side SCAN to discover keys (compatible with Redis client prefix hooks in tests), then passes them to a Lua script for atomic member-level cleanup. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -43,6 +43,9 @@ type ConcurrencyCache interface {
|
||||
|
||||
// 清理过期槽位(后台任务)
|
||||
CleanupExpiredAccountSlots(ctx context.Context, accountID int64) error
|
||||
|
||||
// 启动时清理旧进程遗留槽位与等待计数
|
||||
CleanupStaleProcessSlots(ctx context.Context, activeRequestPrefix string) error
|
||||
}
|
||||
|
||||
var (
|
||||
@@ -59,13 +62,22 @@ func initRequestIDPrefix() string {
|
||||
return "r" + strconv.FormatUint(fallback, 36)
|
||||
}
|
||||
|
||||
// generateRequestID generates a unique request ID for concurrency slot tracking.
|
||||
// Format: {process_random_prefix}-{base36_counter}
|
||||
func RequestIDPrefix() string {
|
||||
return requestIDPrefix
|
||||
}
|
||||
|
||||
func generateRequestID() string {
|
||||
seq := requestIDCounter.Add(1)
|
||||
return requestIDPrefix + "-" + strconv.FormatUint(seq, 36)
|
||||
}
|
||||
|
||||
func (s *ConcurrencyService) CleanupStaleProcessSlots(ctx context.Context) error {
|
||||
if s == nil || s.cache == nil {
|
||||
return nil
|
||||
}
|
||||
return s.cache.CleanupStaleProcessSlots(ctx, RequestIDPrefix())
|
||||
}
|
||||
|
||||
const (
|
||||
// Default extra wait slots beyond concurrency limit
|
||||
defaultExtraWaitSlots = 20
|
||||
|
||||
@@ -91,6 +91,32 @@ func (c *stubConcurrencyCacheForTest) CleanupExpiredAccountSlots(_ context.Conte
|
||||
return c.cleanupErr
|
||||
}
|
||||
|
||||
func (c *stubConcurrencyCacheForTest) CleanupStaleProcessSlots(_ context.Context, _ string) error {
|
||||
return c.cleanupErr
|
||||
}
|
||||
|
||||
type trackingConcurrencyCache struct {
|
||||
stubConcurrencyCacheForTest
|
||||
cleanupPrefix string
|
||||
}
|
||||
|
||||
func (c *trackingConcurrencyCache) CleanupStaleProcessSlots(_ context.Context, prefix string) error {
|
||||
c.cleanupPrefix = prefix
|
||||
return c.cleanupErr
|
||||
}
|
||||
|
||||
func TestCleanupStaleProcessSlots_NilCache(t *testing.T) {
|
||||
svc := &ConcurrencyService{cache: nil}
|
||||
require.NoError(t, svc.CleanupStaleProcessSlots(context.Background()))
|
||||
}
|
||||
|
||||
func TestCleanupStaleProcessSlots_DelegatesPrefix(t *testing.T) {
|
||||
cache := &trackingConcurrencyCache{}
|
||||
svc := NewConcurrencyService(cache)
|
||||
require.NoError(t, svc.CleanupStaleProcessSlots(context.Background()))
|
||||
require.Equal(t, RequestIDPrefix(), cache.cleanupPrefix)
|
||||
}
|
||||
|
||||
func TestAcquireAccountSlot_Success(t *testing.T) {
|
||||
cache := &stubConcurrencyCacheForTest{acquireResult: true}
|
||||
svc := NewConcurrencyService(cache)
|
||||
|
||||
@@ -1986,6 +1986,10 @@ func (m *mockConcurrencyCache) CleanupExpiredAccountSlots(ctx context.Context, a
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *mockConcurrencyCache) CleanupStaleProcessSlots(ctx context.Context, activeRequestPrefix string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *mockConcurrencyCache) GetUsersLoadBatch(ctx context.Context, users []UserWithConcurrency) (map[int64]*UserLoadInfo, error) {
|
||||
result := make(map[int64]*UserLoadInfo, len(users))
|
||||
for _, user := range users {
|
||||
|
||||
@@ -105,6 +105,9 @@ func ProvideDeferredService(accountRepo AccountRepository, timingWheel *TimingWh
|
||||
// ProvideConcurrencyService creates ConcurrencyService and starts slot cleanup worker.
|
||||
func ProvideConcurrencyService(cache ConcurrencyCache, accountRepo AccountRepository, cfg *config.Config) *ConcurrencyService {
|
||||
svc := NewConcurrencyService(cache)
|
||||
if err := svc.CleanupStaleProcessSlots(context.Background()); err != nil {
|
||||
logger.LegacyPrintf("service.concurrency", "Warning: startup cleanup stale process slots failed: %v", err)
|
||||
}
|
||||
if cfg != nil {
|
||||
svc.StartSlotCleanupWorker(accountRepo, cfg.Gateway.Scheduling.SlotCleanupInterval)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user