Merge pull request #3771 from Wei-Shaw/fix/redis-scan-index-hardening

fix(redis-index): 加固 #3762 引入的索引驱动清理架构

REDIS_SCAN_ARCHITECTURE_OPTIMIZATION.md

@@ -1,568 +0,0 @@
-# Redis SCAN 架构优化执行文档
-
-本文是架构级执行文档，只覆盖三个目标：
-
-1. 账号并发活跃负载查询不再通过 Redis keyspace `SCAN` 发现账号。
-2. 账号/用户并发槽过期清理、启动遗留槽清理不再通过 Redis keyspace `SCAN` 发现 key。
-3. 用户消息队列孤儿锁清理不再通过 Redis keyspace `SCAN` 发现 lock key。
-
-不覆盖旁路录制、OpenAI failover、日志量、业务限流配置调参。不要把本文扩展成短期止血方案。
-
-## 成功标准
-
-实现完成后必须同时满足：
-
-- `backend/internal/repository/concurrency_cache.go` 中不得再调用 `c.rdb.Scan(...)`。
-- `backend/internal/repository/user_msg_queue_cache.go` 中不得再调用 `c.rdb.Scan(...)`。
-- `backend/internal/service/user_msg_queue_service.go` 中不得再出现 `ScanLockKeys` 接口调用。
-- `GetActiveAccountLoadMap` 只读显式维护的 Redis 索引，不扫描 Redis keyspace。
-- `CleanupExpiredAccountSlotKeys` 只处理显式索引中的候选账号，不扫描 Redis keyspace。
-- `CleanupStaleProcessSlots` 不扫描 Redis keyspace；它必须基于显式索引清理，或只依赖 TTL/score 自然过期。
-- UMQ cleanup worker 只读 `umq:lock:index` 候选项，不扫描 `umq:{*}:lock`。
-- 主业务并发限制仍以原账号/用户 slot key 为准，不能以索引为准。索引只能用于发现候选对象、监控和清理。
-
-执行完必须用下面命令确认生产代码没有遗留扫描：
-
-```powershell
-rg -n "\.Scan\(" backend/internal/repository/concurrency_cache.go backend/internal/repository/user_msg_queue_cache.go backend/internal/service/user_msg_queue_service.go
-rg -n "ScanLockKeys|scanAccountIDsByPrefix|cleanupSlotsByPattern|deleteKeysByPattern|umqScanPattern" backend/internal/repository/concurrency_cache.go backend/internal/repository/user_msg_queue_cache.go backend/internal/service/user_msg_queue_service.go
-```
-
-第一条必须无输出。第二条必须无生产函数残留；测试文件不在此检查范围。
-
-## 不可违反的约束
-
-- 不要用 `KEYS` 替代 `SCAN`。
-- 不要把全量 Redis keyspace 扫描移动到另一个函数、goroutine、启动流程或管理接口里。
-- 不要在请求路径、后台 worker、启动流程中做 Redis keyspace pattern enumeration。
-- 不要在 Redis Lua 脚本里同时操作“全局索引 key”和“账号/用户局部 key”。项目代码当前有 Redis Cluster 兼容要求，这种写法会在 Cluster 下触发 CROSSSLOT。
-- 索引更新失败不得改变主业务 acquire/release 的成功结果。索引是 best-effort discovery structure，不是并发正确性的来源。
-- 不能因为索引缺失而拒绝用户请求。索引缺失最多影响 Ops 实时视图和后台提前清理；原 slot/wait key 的 TTL 必须保证最终自愈。
-
-## 新增 Redis Key
-
-### 并发索引
-
-在 `backend/internal/repository/concurrency_cache.go` 增加常量：
-
-```go
-const (
-	accountActiveIndexKey = "concurrency:account:active_index" // ZSET member=accountID, score=expireAtUnixSeconds
-	userActiveIndexKey    = "concurrency:user:active_index"    // ZSET member=userID, score=expireAtUnixSeconds
-
-	activeIndexCleanupBatchSize  = 1000
-	activeIndexPipelineChunkSize = 500
-)
-```
-
-语义：
-
-- `accountActiveIndexKey` 记录“可能有账号槽位或账号等待计数”的账号 ID。
-- `userActiveIndexKey` 记录“可能有用户槽位或用户等待计数”的用户 ID。
-- ZSET score 是候选对象的保守过期时间，单位为 Unix 秒。
-- member 必须是十进制 ID 字符串，不要存完整 Redis key。
-- 索引允许短暂 stale；读索引后必须二次查询真实 slot/wait key。
-
-score 规则：
-
-- 成功获取账号槽位：score = Redis 当前秒 + `slotTTLSeconds`。
-- 成功增加账号等待计数：score = Redis 当前秒 + `waitQueueTTLSeconds`。
-- 成功获取用户槽位：score = Redis 当前秒 + `slotTTLSeconds`。
-- 成功增加用户等待计数：score = Redis 当前秒 + `waitQueueTTLSeconds`。
-- release/decrement 后如果真实 slot count 和 wait count 都为 0，则从索引 `ZREM`。
-- release/decrement 后如果仍有 slot 或 wait，则重新 `ZADD` 一个新的保守过期时间。
-
-### UMQ 锁索引
-
-在 `backend/internal/repository/user_msg_queue_cache.go` 增加常量：
-
-```go
-const (
-	umqLockIndexKey = "umq:lock:index" // ZSET member=accountID, score=lockExpireAtUnixMs
-	umqLockIndexCleanupBatchSize = 1000
-)
-```
-
-语义：
-
-- `umqLockIndexKey` 记录“可能存在 UMQ lock”的账号 ID。
-- ZSET score 是 lock 的预计过期时间，单位为 Unix 毫秒。
-- member 必须是十进制 accountID 字符串。
-- 索引只用于 cleanup worker 找候选 lock。锁是否存在、是否孤儿，必须再查 `umq:{accountID}:lock`。
-
-## 第一部分：并发活跃索引
-
-修改文件：`backend/internal/repository/concurrency_cache.go`。
-
-### 1.1 增加 Redis 时间 helper
-
-新增 helper，所有索引 score 使用 Redis server time，不用本机时间：
-
-```go
-func (c *concurrencyCache) redisUnixSeconds(ctx context.Context) (int64, error) {
-	now, err := c.rdb.Time(ctx).Result()
-	if err != nil {
-		return 0, fmt.Errorf("redis TIME: %w", err)
-	}
-	return now.Unix(), nil
-}
-```
-
-不要在 Lua 脚本里写全局索引，避免 CROSSSLOT。
-
-### 1.2 增加索引 touch/remove/refresh helper
-
-新增以下 helper。名字可以微调，但行为不能改。
-
-```go
-func (c *concurrencyCache) touchAccountActiveIndex(ctx context.Context, accountID int64, ttlSeconds int) {
-	c.touchActiveIndex(ctx, accountActiveIndexKey, accountID, ttlSeconds)
-}
-
-func (c *concurrencyCache) touchUserActiveIndex(ctx context.Context, userID int64, ttlSeconds int) {
-	c.touchActiveIndex(ctx, userActiveIndexKey, userID, ttlSeconds)
-}
-
-func (c *concurrencyCache) touchActiveIndex(ctx context.Context, indexKey string, id int64, ttlSeconds int) {
-	if c == nil || c.rdb == nil || id <= 0 || ttlSeconds <= 0 {
-		return
-	}
-	now, err := c.redisUnixSeconds(ctx)
-	if err != nil {
-		return
-	}
-	_ = c.rdb.ZAdd(ctx, indexKey, redis.Z{
-		Score:  float64(now + int64(ttlSeconds)),
-		Member: strconv.FormatInt(id, 10),
-	}).Err()
-}
-```
-
-索引维护是 best-effort，所以 helper 内部吞掉错误。不要把索引错误返回给 acquire/release 调用方。
-
-再新增 refresh helper：
-
-```go
-func (c *concurrencyCache) refreshAccountActiveIndex(ctx context.Context, accountID int64) {
-	// 真实状态以 accountSlotKey(accountID) 和 accountWaitKey(accountID) 为准。
-	// 先清理该账号 slot 中过期成员，再读 ZCARD 和 GET wait。
-	// 如果 slotCount == 0 && waitCount <= 0：ZREM accountActiveIndexKey accountID。
-	// 否则：ZADD accountActiveIndexKey accountID，score = now + maxRelevantTTL。
-}
-
-func (c *concurrencyCache) refreshUserActiveIndex(ctx context.Context, userID int64) {
-	// 真实状态以 userSlotKey(userID) 和 waitQueueKey(userID) 为准。
-	// 行为同 refreshAccountActiveIndex。
-}
-```
-
-实现要求：
-
-- `refresh*` 必须 best-effort，不能向 release/decrement 返回索引错误。
-- `waitCount` 读取 `redis.Nil` 时按 0 处理。
-- `waitCount < 0` 必须按 0 处理。
-- `slotCount > 0` 时 score 至少延长 `slotTTLSeconds`。
-- `waitCount > 0` 时 score 至少延长 `waitQueueTTLSeconds`。
-- 两者都存在时使用更大的 TTL。
-
-### 1.3 修改账号写路径
-
-修改 `AcquireAccountSlot`：
-
-```go
-result, err := acquireScript.Run(...).Int()
-if err != nil { return false, err }
-if result == 1 {
-	c.touchAccountActiveIndex(ctx, accountID, c.slotTTLSeconds)
-}
-return result == 1, nil
-```
-
-修改 `ReleaseAccountSlot`：
-
-```go
-if err := c.rdb.ZRem(ctx, key, requestID).Err(); err != nil {
-	return err
-}
-c.refreshAccountActiveIndex(ctx, accountID)
-return nil
-```
-
-修改 `IncrementAccountWaitCount`：
-
-```go
-result, err := incrementAccountWaitScript.Run(...).Int()
-if err != nil { return false, err }
-if result == 1 {
-	c.touchAccountActiveIndex(ctx, accountID, c.waitQueueTTLSeconds)
-}
-return result == 1, nil
-```
-
-修改 `DecrementAccountWaitCount`：
-
-```go
-_, err := decrementWaitScript.Run(...).Result()
-if err == nil {
-	c.refreshAccountActiveIndex(ctx, accountID)
-}
-return err
-```
-
-### 1.4 修改用户写路径
-
-同账号路径，修改：
-
-- `AcquireUserSlot`
-- `ReleaseUserSlot`
-- `IncrementWaitCount`
-- `DecrementWaitCount`
-
-用户索引使用 `userActiveIndexKey`。
-
-### 1.5 重写 GetActiveAccountLoadMap
-
-删除 `scanAccountIDsByPrefix` 和 `parseAccountIDFromPrefixedKey` 的生产调用。`GetActiveAccountLoadMap` 必须改成：
-
-1. 获取 Redis 当前秒。
-2. `ZRemRangeByScore(accountActiveIndexKey, "-inf", strconv.FormatInt(now, 10))` 删除过期候选。
-3. `ZRangeByScore(accountActiveIndexKey, &redis.ZRangeBy{Min: strconv.FormatInt(now+1, 10), Max: "+inf"})` 获取候选账号 ID。
-4. 解析 member 为 `int64`，非法 member 记录到待删除列表。
-5. 分块 pipeline，块大小 `activeIndexPipelineChunkSize`。
-6. 对每个候选账号执行：
-   - `ZRemRangeByScore(accountSlotKey(id), "-inf", cutoffUnixSeconds)`
-   - `ZCard(accountSlotKey(id))`
-   - `Get(accountWaitKey(id))`
-7. 构造结果时只返回 `currentConcurrency > 0 || waitingCount > 0` 的账号。
-8. 对真实状态为空或 member 非法的账号执行 `ZREM accountActiveIndexKey member`。
-9. 对真实状态仍活跃但 index score 已接近过期的账号，调用 `touchAccountActiveIndex` 刷新。
-
-禁止：
-
-- 禁止再扫 `concurrency:account:*`。
-- 禁止再扫 `wait:account:*`。
-- 禁止用索引里的 score 直接判断并发数。
-
-## 第二部分：并发槽清理和启动清理
-
-修改文件：`backend/internal/repository/concurrency_cache.go`。
-
-### 2.1 重写 CleanupExpiredAccountSlotKeys
-
-当前实现调用 `cleanupExpiredSlotKeysByPattern(ctx, accountSlotKeyPrefix+"*")`，必须删除。
-
-新行为：
-
-1. 获取 Redis 当前秒 `now`。
-2. 从 `accountActiveIndexKey` 读取过期候选：
-
-```go
-ids, err := c.rdb.ZRangeByScore(ctx, accountActiveIndexKey, &redis.ZRangeBy{
-	Min: "-inf",
-	Max: strconv.FormatInt(now, 10),
-	Count: activeIndexCleanupBatchSize,
-}).Result()
-```
-
-3. 对每个候选账号清理该账号 slot 过期成员并读真实状态。
-4. 如果真实 `slotCount == 0 && waitCount <= 0`，从 `accountActiveIndexKey` 删除该账号。
-5. 如果真实仍活跃，刷新 `accountActiveIndexKey` score。
-6. 不需要处理不在索引中的账号；其 slot key 自身有 `EXPIRE`，并且 acquire/get-load 会惰性清理过期成员。
-
-这个函数不再表示“遍历所有账号槽位 key”，而是“处理索引中到期的账号候选”。保留原函数名是为了少改接口。
-
-### 2.2 重写 CleanupStaleProcessSlots
-
-当前实现会扫描：
-
-- `concurrency:account:*`
-- `concurrency:user:*`
-- `wait:account:*`
-- `concurrency:wait:*`
-
-必须去掉这些扫描。
-
-新行为必须基于索引：
-
-1. 从 `accountActiveIndexKey` 读取所有未过期候选账号。
-2. 对每个账号：
-   - 对 `accountSlotKey(id)` 运行“单 key 清理脚本”，删除 requestID 前缀不是当前 `activeRequestPrefix` 的成员。
-   - 删除 `accountWaitKey(id)`，因为等待者属于旧进程，重启后不能继续等待。
-   - 调用 `refreshAccountActiveIndex(ctx, id)`。
-3. 从 `userActiveIndexKey` 读取所有未过期候选用户。
-4. 对每个用户：
-   - 对 `userSlotKey(id)` 运行同一个“单 key 清理脚本”。
-   - 删除 `waitQueueKey(id)`。
-   - 调用 `refreshUserActiveIndex(ctx, id)`。
-
-新增单 key Lua 脚本，替代当前 `startupCleanupScript` 的多 key 版本：
-
-```lua
-local key = KEYS[1]
-local activePrefix = ARGV[1]
-local slotTTL = tonumber(ARGV[2])
-local removed = 0
-local members = redis.call('ZRANGE', key, 0, -1)
-for _, member in ipairs(members) do
-  if string.sub(member, 1, string.len(activePrefix)) ~= activePrefix then
-    removed = removed + redis.call('ZREM', key, member)
-  end
-end
-if redis.call('ZCARD', key) == 0 then
-  redis.call('DEL', key)
-else
-  redis.call('EXPIRE', key, slotTTL)
-end
-return removed
-```
-
-该脚本只接受一个 slot key，避免 Redis Cluster CROSSSLOT。
-
-如果索引不存在或为空：
-
-- `CleanupStaleProcessSlots` 直接返回 nil。
-- 不要 fallback 到 `SCAN`。
-- 旧版本遗留 key 依赖 Redis TTL 自然过期。不要在 app 启动时做兼容性 keyspace backfill。
-
-### 2.3 删除旧扫描函数
-
-删除以下生产函数：
-
-- `scanAccountIDsByPrefix`
-- `parseAccountIDFromPrefixedKey`，如果没有其他生产调用
-- `cleanupExpiredSlotKeysByPattern`
-- `cleanupSlotsByPattern`
-- `deleteKeysByPattern`
-
-如果测试需要解析 key，测试内自建 helper，不要保留生产 helper。
-
-## 第三部分：UMQ 锁索引
-
-修改文件：
-
-- `backend/internal/repository/user_msg_queue_cache.go`
-- `backend/internal/service/user_msg_queue_service.go`
-
-### 3.1 修改 service 接口
-
-在 `backend/internal/service/user_msg_queue_service.go` 的 `UserMsgQueueCache` 接口中删除：
-
-```go
-ScanLockKeys(ctx context.Context, maxCount int) ([]int64, error)
-ForceReleaseLock(ctx context.Context, accountID int64) error
-```
-
-替换为：
-
-```go
-ReconcileExpiredLockCandidates(ctx context.Context, maxCount int) (cleaned int, err error)
-```
-
-原因：cleanup worker 不应该知道 Redis lock key 的枚举方式，也不应该先枚举再逐个 `ForceReleaseLock`。候选读取、PTTL 校验、索引刷新应该封装在 cache 层。
-
-### 3.2 修改 acquireLockScript 返回值
-
-当前脚本只返回 0/1。改成返回数组：
-
-```lua
-redis.replicate_commands()
-local cur = redis.call('GET', KEYS[1])
-local ttl = tonumber(ARGV[2])
-if cur == ARGV[1] then
-  redis.call('PEXPIRE', KEYS[1], ttl)
-  local t = redis.call('TIME')
-  local ms = tonumber(t[1])*1000 + math.floor(tonumber(t[2])/1000)
-  return {1, ms + ttl}
-end
-if cur ~= false then
-  return {0, 0}
-end
-redis.call('SET', KEYS[1], ARGV[1], 'PX', ttl)
-local t = redis.call('TIME')
-local ms = tonumber(t[1])*1000 + math.floor(tonumber(t[2])/1000)
-return {1, ms + ttl}
-```
-
-Go 侧解析：
-
-- 第一个元素是 acquired，1 表示拿到锁。
-- 第二个元素是 Redis 时间计算出的 `expireAtUnixMs`。
-- acquired 为 1 时，best-effort 写 `ZADD umqLockIndexKey expireAtMs accountID`。
-- `ZADD` 失败不能让 `AcquireLock` 返回失败。
-
-### 3.3 修改 ReleaseLock
-
-`ReleaseLock` 主逻辑保持原子释放锁和写 last key。
-
-释放成功时：
-
-```go
-if result == 1 {
-	_ = c.rdb.ZRem(ctx, umqLockIndexKey, strconv.FormatInt(accountID, 10)).Err()
-}
-```
-
-释放失败时不要删除索引。失败可能是 requestID 不匹配或 lock 已过期；cleanup worker 会处理 stale index。
-
-### 3.4 新增 reconcile 脚本
-
-删除 `forceReleaseLockScript` 的外部使用。新增脚本：
-
-```lua
-local pttl = redis.call('PTTL', KEYS[1])
-if pttl == -2 then
-  return {-2, 0}
-end
-if pttl == -1 then
-  redis.call('DEL', KEYS[1])
-  return {-1, 0}
-end
-return {1, pttl}
-```
-
-返回语义：
-
-- `-2`：lock key 不存在。Go 侧 `ZREM umqLockIndexKey accountID`。
-- `-1`：lock key 存在但无 TTL，脚本已删除。Go 侧 `ZREM umqLockIndexKey accountID`，cleaned++。
-- `1`：lock key 仍有 TTL。Go 侧用 Redis 当前毫秒 + pttl 刷新 `umqLockIndexKey` score。
-
-### 3.5 实现 ReconcileExpiredLockCandidates
-
-实现步骤：
-
-1. 用 `c.rdb.Time(ctx)` 获取 Redis 当前毫秒 `nowMs`。
-2. 从 `umqLockIndexKey` 取到期候选：
-
-```go
-members, err := c.rdb.ZRangeByScore(ctx, umqLockIndexKey, &redis.ZRangeBy{
-	Min: "-inf",
-	Max: strconv.FormatInt(nowMs, 10),
-	Count: int64(maxCount),
-}).Result()
-```
-
-3. 逐个解析 accountID。非法 member 直接 `ZREM`。
-4. 对合法 accountID 运行 reconcile 脚本，key 为 `umqLockKey(accountID)`。
-5. 根据返回值删除索引、刷新索引或累计 cleaned。
-6. 函数返回 cleaned 数。
-
-禁止：
-
-- 禁止 fallback 到 `SCAN umq:{*}:lock`。
-- 禁止用 `KEYS umq:*`。
-- 禁止 cleanup worker 自己解析 lock key。
-
-### 3.6 修改 StartCleanupWorker
-
-当前 worker 先 `ScanLockKeys` 再逐个 `ForceReleaseLock`。改成：
-
-```go
-cleaned, err := s.cache.ReconcileExpiredLockCandidates(ctx, 1000)
-if err != nil {
-	logger.LegacyPrintf("service.umq", "Cleanup reconcile failed: %v", err)
-	return
-}
-if cleaned > 0 {
-	logger.LegacyPrintf("service.umq", "Cleanup completed: released %d orphaned locks", cleaned)
-}
-```
-
-worker 不再知道扫描、PTTL、索引等细节。
-
-### 3.7 删除旧 UMQ 扫描函数
-
-删除：
-
-- `umqScanPattern`
-- `ScanLockKeys`
-- `ForceReleaseLock`，如果无生产调用
-
-如果测试仍需要强造 PTTL == -1 的 key，只在测试里直接写 Redis。
-
-## 测试要求
-
-### 并发缓存测试
-
-新增或修改 `backend/internal/repository/concurrency_cache_*_test.go`。
-
-必须覆盖：
-
-1. `AcquireAccountSlot` 成功后 `GetActiveAccountLoadMap` 能看到该账号。
-2. `ReleaseAccountSlot` 后 `GetActiveAccountLoadMap` 不再返回该账号。
-3. `IncrementAccountWaitCount` 成功后 `GetActiveAccountLoadMap` 能看到 waiting count。
-4. `DecrementAccountWaitCount` 后如果无 slot，则索引被移除。
-5. `CleanupExpiredAccountSlotKeys` 不依赖 keyspace scan：测试里只创建索引成员和对应 slot key，然后确认会清理；再创建未索引 slot key，确认不会被该函数主动发现。
-6. `CleanupStaleProcessSlots` 只处理索引中的 account/user，删除旧 request prefix 成员，保留当前 prefix 成员，删除 account/user wait key。
-7. 索引中存在非法 member 时，`GetActiveAccountLoadMap` 不报错，并移除非法 member。
-
-### UMQ 测试
-
-新增或修改 `backend/internal/repository/user_msg_queue_cache*_test.go` 和 `backend/internal/service/user_msg_queue_service*_test.go`。
-
-必须覆盖：
-
-1. `AcquireLock` 成功后写入 `umq:lock:index`，score 大于 Redis 当前毫秒。
-2. `ReleaseLock` 成功后删除 `umq:lock:index` member。
-3. lock 已自然过期时，`ReconcileExpiredLockCandidates` 删除 stale index member。
-4. lock 仍有 TTL 但 index score 到期时，`ReconcileExpiredLockCandidates` 刷新 index score，不删除 lock。
-5. lock 存在且 `PTTL == -1` 时，`ReconcileExpiredLockCandidates` 删除 lock，删除 index member，并返回 cleaned=1。
-6. index 中非法 member 不导致错误，并被删除。
-7. `StartCleanupWorker` 调用 `ReconcileExpiredLockCandidates`，不再调用 `ScanLockKeys` 或 `ForceReleaseLock`。
-
-### 禁止项测试
-
-实现完成后运行：
-
-```powershell
-rg -n "\.Scan\(" backend/internal/repository/concurrency_cache.go backend/internal/repository/user_msg_queue_cache.go backend/internal/service/user_msg_queue_service.go
-rg -n "ScanLockKeys|umqScanPattern|scanAccountIDsByPrefix|cleanupExpiredSlotKeysByPattern|cleanupSlotsByPattern|deleteKeysByPattern" backend/internal/repository/concurrency_cache.go backend/internal/repository/user_msg_queue_cache.go backend/internal/service/user_msg_queue_service.go
-```
-
-上述命令必须无输出。
-
-再运行相关测试。按项目约定，编译很慢时先把代码复制到 WSL 文件系统再跑：
-
-```bash
-cd backend
-go test ./internal/repository ./internal/service
-```
-
-如果全量包太慢，至少先跑：
-
-```bash
-cd backend
-go test ./internal/repository -run 'Concurrency|UserMsgQueue|Redis'
-go test ./internal/service -run 'Concurrency|UserMessageQueue'
-```
-
-## 迁移和兼容
-
-不要在应用启动时扫描旧 key 回填索引。
-
-原因：
-
-- 这会把问题从运行期 `SCAN` 搬到启动期 `SCAN`。
-- 生产实例重启时 Redis 已经高 CPU，启动扫描会放大抖动。
-- 并发 slot key 和 wait key 都有 TTL，新版本写路径会为新流量维护索引，旧 key 可自然过期。
-
-兼容策略：
-
-- 新版本上线后，新请求会逐步填充 `concurrency:*:active_index` 和 `umq:lock:index`。
-- 旧并发 slot key 没有索引时，不影响并发限制本身；对应账号下一次 acquire/get-load 会清理自己的 slot。
-- 旧 UMQ lock 如果有 TTL，会自然过期。
-- 极少数历史 `PTTL == -1` UMQ lock 且没有 index 的情况，不由应用自动发现。需要人工离线维护时，单独写一次性脚本，维护窗口运行，不要放进服务启动或后台 worker。
-
-## 代码审查检查表
-
-提交前逐项确认：
-
-- [ ] 没有新增 `KEYS`。
-- [ ] 没有新增生产路径 `SCAN`。
-- [ ] 没有在 Lua 脚本中同时操作全局索引 key 和账号/用户局部 key。
-- [ ] 索引维护失败不会让 acquire/release/decrement 的主结果失败。
-- [ ] `GetActiveAccountLoadMap` 对 stale index、非法 member、Redis nil 都能正常返回。
-- [ ] `CleanupExpiredAccountSlotKeys` 不再遍历 keyspace。
-- [ ] `CleanupStaleProcessSlots` 不再遍历 keyspace。
-- [ ] UMQ cleanup worker 不再知道 lock key pattern。
-- [ ] 所有旧扫描 helper 已删除或仅存在于测试文件。
-- [ ] 新测试覆盖成功路径、stale index、非法 member、PTTL -1、自然过期。

backend/internal/repository/concurrency_cache.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"strconv"
 
+	"github.com/Wei-Shaw/sub2api/internal/pkg/logger"
 	"github.com/Wei-Shaw/sub2api/internal/service"
 	"github.com/redis/go-redis/v9"
 )
@@ -45,6 +46,10 @@ const (
 	// 后台清理只按批处理索引候选，避免单次任务占用 Redis 太久。
 	activeIndexCleanupBatchSize  = 1000
 	activeIndexPipelineChunkSize = 500
+
+	// 一次性迁移 marker：活跃索引机制上线前遗留的等待计数键无法被索引发现，
+	// 且有流量时 TTL 会被不断刷新，必须清扫一次。marker 存在即代表已完成。
+	legacyWaitSweepMarkerKey = "concurrency:startup:legacy_wait_sweep:v1"
 )
 
 var (
@@ -54,6 +59,7 @@ var (
 	// ARGV[1] = maxConcurrency
 	// ARGV[2] = TTL（秒）
 	// ARGV[3] = requestID
+	// 返回 {是否成功, Redis 当前秒}，Go 侧复用同一时间源写活跃索引，省去额外 TIME 往返。
 	acquireScript = 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).
@@ -76,7 +82,7 @@ var (
 		if exists ~= false then
 			redis.call('ZADD', key, now, requestID)
 			redis.call('EXPIRE', key, ttl)
-			return 1
+			return {1, now}
 		end
 
 		-- 检查是否达到并发上限
@@ -84,10 +90,10 @@ var (
 		if count < maxConcurrency then
 			redis.call('ZADD', key, now, requestID)
 			redis.call('EXPIRE', key, ttl)
-			return 1
+			return {1, now}
 		end
 
-		return 0
+		return {0, now}
 	`)
 
 	// getCountScript 统计有序集合中的槽位数量并清理过期条目
@@ -136,46 +142,56 @@ var (
 	// KEYS[1] = wait queue key
 	// ARGV[1] = maxWait
 	// ARGV[2] = TTL in seconds
+	// 返回 {是否成功, Redis 当前秒}，供 Go 侧免额外 TIME 往返写活跃索引。
 	incrementWaitScript = 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 current = redis.call('GET', KEYS[1])
 		if current == false then
 			current = 0
 		else
 			current = tonumber(current)
 		end
+		local now = tonumber(redis.call('TIME')[1])
 
 		if current >= tonumber(ARGV[1]) then
-			return 0
+			return {0, now}
 		end
 
-		local newVal = redis.call('INCR', KEYS[1])
+		redis.call('INCR', KEYS[1])
 
 		-- Refresh TTL so long-running traffic doesn't expire active queue counters.
 		redis.call('EXPIRE', KEYS[1], ARGV[2])
 
-			return 1
-		`)
+		return {1, now}
+	`)
 
 	// incrementAccountWaitScript - account-level wait queue count (refresh TTL on each increment)
+	// 返回值同 incrementWaitScript：{是否成功, Redis 当前秒}。
 	incrementAccountWaitScript = redis.NewScript(`
-			local current = redis.call('GET', KEYS[1])
-			if current == false then
-				current = 0
-			else
-				current = tonumber(current)
-			end
+		-- 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 current = redis.call('GET', KEYS[1])
+		if current == false then
+			current = 0
+		else
+			current = tonumber(current)
+		end
+		local now = tonumber(redis.call('TIME')[1])
 
-			if current >= tonumber(ARGV[1]) then
-				return 0
-			end
+		if current >= tonumber(ARGV[1]) then
+			return {0, now}
+		end
 
-			local newVal = redis.call('INCR', KEYS[1])
+		redis.call('INCR', KEYS[1])
 
-			-- Refresh TTL so long-running traffic doesn't expire active queue counters.
-			redis.call('EXPIRE', KEYS[1], ARGV[2])
+		-- Refresh TTL so long-running traffic doesn't expire active queue counters.
+		redis.call('EXPIRE', KEYS[1], ARGV[2])
 
-			return 1
-		`)
+		return {1, now}
+	`)
 
 	// decrementWaitScript - same as before
 	decrementWaitScript = redis.NewScript(`
@@ -209,6 +225,7 @@ var (
 
 	// startupCleanupSlotScript 清理单个槽位 key 中非当前进程前缀的成员，避免 Redis Cluster CROSSSLOT。
 	// KEYS[1] 是有序集合键，ARGV[1] 是当前进程前缀，ARGV[2] 是槽位 TTL。
+	// 返回 {清除数量, 剩余成员数}，Go 侧据剩余数决定索引 member 去留，无需再回读槽位。
 	startupCleanupSlotScript = redis.NewScript(`
 		local key = KEYS[1]
 		local activePrefix = ARGV[1]
@@ -220,12 +237,13 @@ var (
 				removed = removed + redis.call('ZREM', key, member)
 			end
 		end
-		if redis.call('ZCARD', key) == 0 then
+		local remaining = redis.call('ZCARD', key)
+		if remaining == 0 then
 			redis.call('DEL', key)
 		else
 			redis.call('EXPIRE', key, slotTTL)
 		end
-		return removed
+		return {removed, remaining}
 	`)
 )
 
@@ -282,28 +300,32 @@ func (c *concurrencyCache) redisUnixSeconds(ctx context.Context) (int64, error)
 	return now.Unix(), nil
 }
 
-func (c *concurrencyCache) touchAccountActiveIndex(ctx context.Context, accountID int64, ttlSeconds int) {
-	c.touchActiveIndex(ctx, accountActiveIndexKey, accountID, ttlSeconds)
+// slotIndexSpec 描述一个活跃索引及其对应的槽位/等待键构造方式。
+// 用具名字段避免把 slotKey/waitKey 两个同签名函数按位置传参时写反。
+type slotIndexSpec struct {
+	indexKey string
+	slotKey  func(int64) string
+	waitKey  func(int64) string
 }
 
-func (c *concurrencyCache) touchUserActiveIndex(ctx context.Context, userID int64, ttlSeconds int) {
-	c.touchActiveIndex(ctx, userActiveIndexKey, userID, ttlSeconds)
-}
+var (
+	accountSlotIndex = slotIndexSpec{indexKey: accountActiveIndexKey, slotKey: accountSlotKey, waitKey: accountWaitKey}
+	userSlotIndex    = slotIndexSpec{indexKey: userActiveIndexKey, slotKey: userSlotKey, waitKey: waitQueueKey}
+)
 
-// touchActiveIndex 是写路径上的轻量标记：主操作已成功时，尽力把 ID 放入活跃索引。
-// 索引失败不影响并发槽位/等待队列本身，后续释放或清理会再次校正。
-func (c *concurrencyCache) touchActiveIndex(ctx context.Context, indexKey string, id int64, ttlSeconds int) {
-	if c == nil || c.rdb == nil || id <= 0 || ttlSeconds <= 0 {
+// touchActiveIndexAt 是写路径上的轻量标记：主操作已成功时，尽力把 ID 放入活跃索引，
+// score 为给定的绝对过期时间（Redis Unix 秒）。索引失败不影响并发槽位/等待队列本身，
+// 后续释放或清理会再次校正，因此只记日志不上抛。
+func (c *concurrencyCache) touchActiveIndexAt(ctx context.Context, indexKey string, id int64, expireAt int64) {
+	if c == nil || c.rdb == nil || id <= 0 || expireAt <= 0 {
 		return
 	}
-	now, err := c.redisUnixSeconds(ctx)
-	if err != nil {
-		return
-	}
-	_ = c.rdb.ZAdd(ctx, indexKey, redis.Z{
-		Score:  float64(now + int64(ttlSeconds)),
+	if err := c.rdb.ZAdd(ctx, indexKey, redis.Z{
+		Score:  float64(expireAt),
 		Member: strconv.FormatInt(id, 10),
-	}).Err()
+	}).Err(); err != nil {
+		logger.LegacyPrintf("repository.concurrency", "Warning: touch active index %s for %d failed: %v", indexKey, id, err)
+	}
 }
 
 func (c *concurrencyCache) refreshAccountActiveIndex(ctx context.Context, accountID int64) {
@@ -316,22 +338,27 @@ func (c *concurrencyCache) refreshUserActiveIndex(ctx context.Context, userID in
 
 // refreshActiveIndex 以 Redis 中的真实槽位/等待数为准重建索引状态。
 // 释放槽位、等待计数减少、清理过期成员后都会调用它，防止索引残留。
+// 索引维护是 best-effort：失败只记日志，不影响主流程。
 func (c *concurrencyCache) refreshActiveIndex(ctx context.Context, indexKey string, id int64, slotKey, waitKey string) {
 	if c == nil || c.rdb == nil || id <= 0 {
 		return
 	}
 	now, err := c.redisUnixSeconds(ctx)
 	if err != nil {
+		logger.LegacyPrintf("repository.concurrency", "Warning: refresh active index %s for %d failed: %v", indexKey, id, err)
 		return
 	}
 
 	load, err := c.readActiveLoadForKey(ctx, id, slotKey, waitKey, now)
 	if err != nil {
+		logger.LegacyPrintf("repository.concurrency", "Warning: refresh active index %s for %d failed: %v", indexKey, id, err)
 		return
 	}
 	member := strconv.FormatInt(id, 10)
 	if load.slotCount == 0 && load.waitCount <= 0 {
-		_ = c.rdb.ZRem(ctx, indexKey, member).Err()
+		if err := c.rdb.ZRem(ctx, indexKey, member).Err(); err != nil {
+			logger.LegacyPrintf("repository.concurrency", "Warning: remove active index member %s from %s failed: %v", member, indexKey, err)
+		}
 		return
 	}
 
@@ -339,10 +366,7 @@ func (c *concurrencyCache) refreshActiveIndex(ctx context.Context, indexKey stri
 	if ttlSeconds <= 0 {
 		return
 	}
-	_ = c.rdb.ZAdd(ctx, indexKey, redis.Z{
-		Score:  float64(now + int64(ttlSeconds)),
-		Member: member,
-	}).Err()
+	c.touchActiveIndexAt(ctx, indexKey, id, now+int64(ttlSeconds))
 }
 
 type activeIndexLoad struct {
@@ -388,9 +412,9 @@ func (c *concurrencyCache) readActiveLoadForKey(ctx context.Context, id int64, s
 	}, nil
 }
 
-// readAccountIndexLoads 批量读取账号索引候选的真实负载。
+// readIndexLoads 批量读取索引候选的真实负载（账号/用户通用）。
 // 分块 Pipeline 可以减少 Redis 往返，同时避免一次 Pipeline 塞入过多命令。
-func (c *concurrencyCache) readAccountIndexLoads(ctx context.Context, members []string, now int64) ([]activeIndexLoad, []string, error) {
+func (c *concurrencyCache) readIndexLoads(ctx context.Context, spec slotIndexSpec, members []string, now int64) ([]activeIndexLoad, []string, error) {
 	loads := make([]activeIndexLoad, 0, len(members))
 	staleMembers := make([]string, 0)
 	candidates := make([]activeIndexLoad, 0, len(members))
@@ -412,17 +436,17 @@ func (c *concurrencyCache) readAccountIndexLoads(ctx context.Context, members []
 		chunk := candidates[start:end]
 
 		pipe := c.rdb.Pipeline()
-		type accountCmd struct {
+		type loadCmd struct {
 			activeIndexLoad
 			zcardCmd *redis.IntCmd
 			getCmd   *redis.StringCmd
 		}
-		cmds := make([]accountCmd, 0, len(chunk))
+		cmds := make([]loadCmd, 0, len(chunk))
 		for _, candidate := range chunk {
-			slotKey := accountSlotKey(candidate.id)
-			waitKey := accountWaitKey(candidate.id)
+			slotKey := spec.slotKey(candidate.id)
+			waitKey := spec.waitKey(candidate.id)
 			pipe.ZRemRangeByScore(ctx, slotKey, "-inf", strconv.FormatInt(cutoffTime, 10))
-			cmds = append(cmds, accountCmd{
+			cmds = append(cmds, loadCmd{
 				activeIndexLoad: candidate,
 				zcardCmd:        pipe.ZCard(ctx, slotKey),
 				getCmd:          pipe.Get(ctx, waitKey),
@@ -457,12 +481,26 @@ func (c *concurrencyCache) removeActiveIndexMembers(ctx context.Context, indexKe
 	for _, member := range members {
 		args = append(args, member)
 	}
-	_ = c.rdb.ZRem(ctx, indexKey, args...).Err()
+	if err := c.rdb.ZRem(ctx, indexKey, args...).Err(); err != nil {
+		logger.LegacyPrintf("repository.concurrency", "Warning: remove %d active index members from %s failed: %v", len(members), indexKey, err)
+	}
 }
 
-// touchActiveIndexForLoad 根据已读取的真实负载刷新索引过期时间。
-func (c *concurrencyCache) touchActiveIndexForLoad(ctx context.Context, indexKey string, load activeIndexLoad) {
-	c.touchActiveIndex(ctx, indexKey, load.id, c.activeIndexTTL(load.slotCount, load.waitCount))
+// runScriptInt64Pair 执行返回两元素整数数组的 Lua 脚本并解析（如 {result, now}、{removed, remaining}）。
+func runScriptInt64Pair(ctx context.Context, rdb *redis.Client, script *redis.Script, keys []string, args ...any) (int64, int64, error) {
+	raw, err := script.Run(ctx, rdb, keys, args...).Result()
+	if err != nil {
+		return 0, 0, err
+	}
+	first, err := redisScriptInt64At(raw, 0)
+	if err != nil {
+		return 0, 0, fmt.Errorf("parse script value 0: %w", err)
+	}
+	second, err := redisScriptInt64At(raw, 1)
+	if err != nil {
+		return 0, 0, fmt.Errorf("parse script value 1: %w", err)
+	}
+	return first, second, nil
 }
 
 // Account slot operations
@@ -470,13 +508,13 @@ func (c *concurrencyCache) touchActiveIndexForLoad(ctx context.Context, indexKey
 func (c *concurrencyCache) AcquireAccountSlot(ctx context.Context, accountID int64, maxConcurrency int, requestID string) (bool, error) {
 	key := accountSlotKey(accountID)
 	// 时间戳在 Lua 脚本内使用 Redis TIME 命令获取，确保多实例时钟一致
-	result, err := acquireScript.Run(ctx, c.rdb, []string{key}, maxConcurrency, c.slotTTLSeconds, requestID).Int()
+	result, now, err := runScriptInt64Pair(ctx, c.rdb, acquireScript, []string{key}, maxConcurrency, c.slotTTLSeconds, requestID)
 	if err != nil {
 		return false, err
 	}
 	if result == 1 {
 		// 成功占槽后标记活跃账号，后台清理即可从索引定位候选账号。
-		c.touchAccountActiveIndex(ctx, accountID, c.slotTTLSeconds)
+		c.touchActiveIndexAt(ctx, accountActiveIndexKey, accountID, now+int64(c.slotTTLSeconds))
 	}
 	return result == 1, nil
 }
@@ -543,13 +581,13 @@ func (c *concurrencyCache) GetAccountConcurrencyBatch(ctx context.Context, accou
 func (c *concurrencyCache) AcquireUserSlot(ctx context.Context, userID int64, maxConcurrency int, requestID string) (bool, error) {
 	key := userSlotKey(userID)
 	// 时间戳在 Lua 脚本内使用 Redis TIME 命令获取，确保多实例时钟一致
-	result, err := acquireScript.Run(ctx, c.rdb, []string{key}, maxConcurrency, c.slotTTLSeconds, requestID).Int()
+	result, now, err := runScriptInt64Pair(ctx, c.rdb, acquireScript, []string{key}, maxConcurrency, c.slotTTLSeconds, requestID)
 	if err != nil {
 		return false, err
 	}
 	if result == 1 {
 		// 成功占槽后标记活跃用户，避免启动清理依赖全量 SCAN。
-		c.touchUserActiveIndex(ctx, userID, c.slotTTLSeconds)
+		c.touchActiveIndexAt(ctx, userActiveIndexKey, userID, now+int64(c.slotTTLSeconds))
 	}
 	return result == 1, nil
 }
@@ -626,13 +664,13 @@ func (c *concurrencyCache) GetAPIKeyConcurrencyBatch(ctx context.Context, apiKey
 
 func (c *concurrencyCache) IncrementWaitCount(ctx context.Context, userID int64, maxWait int) (bool, error) {
 	key := waitQueueKey(userID)
-	result, err := incrementWaitScript.Run(ctx, c.rdb, []string{key}, maxWait, c.waitQueueTTLSeconds).Int()
+	result, now, err := runScriptInt64Pair(ctx, c.rdb, incrementWaitScript, []string{key}, maxWait, c.waitQueueTTLSeconds)
 	if err != nil {
 		return false, err
 	}
 	if result == 1 {
 		// 等待队列也会让用户保持“活跃”，否则槽位为 0 时后台任务可能漏看等待计数。
-		c.touchUserActiveIndex(ctx, userID, c.waitQueueTTLSeconds)
+		c.touchActiveIndexAt(ctx, userActiveIndexKey, userID, now+int64(c.waitQueueTTLSeconds))
 	}
 	return result == 1, nil
 }
@@ -651,13 +689,13 @@ func (c *concurrencyCache) DecrementWaitCount(ctx context.Context, userID int64)
 
 func (c *concurrencyCache) IncrementAccountWaitCount(ctx context.Context, accountID int64, maxWait int) (bool, error) {
 	key := accountWaitKey(accountID)
-	result, err := incrementAccountWaitScript.Run(ctx, c.rdb, []string{key}, maxWait, c.waitQueueTTLSeconds).Int()
+	result, now, err := runScriptInt64Pair(ctx, c.rdb, incrementAccountWaitScript, []string{key}, maxWait, c.waitQueueTTLSeconds)
 	if err != nil {
 		return false, err
 	}
 	if result == 1 {
 		// 账号级等待队列同样写入账号活跃索引，供负载查询和清理任务使用。
-		c.touchAccountActiveIndex(ctx, accountID, c.waitQueueTTLSeconds)
+		c.touchActiveIndexAt(ctx, accountActiveIndexKey, accountID, now+int64(c.waitQueueTTLSeconds))
 	}
 	return result == 1, nil
 }
@@ -815,113 +853,129 @@ func (c *concurrencyCache) CleanupExpiredAccountSlots(ctx context.Context, accou
 	return err
 }
 
-// GetActiveAccountLoadMap 只读取活跃账号索引中的账号负载。
-// 这是给热路径使用的轻量视图，避免为获取全局账号负载而扫描所有槽位键。
-func (c *concurrencyCache) GetActiveAccountLoadMap(ctx context.Context) (map[int64]*service.AccountLoadInfo, error) {
-	now, err := c.redisUnixSeconds(ctx)
-	if err != nil {
-		return nil, err
+// CleanupExpiredAccountSlotKeys 处理账号与用户两个活跃索引中已到期的候选。
+// （方法名中的 Account 是历史遗留，保留以避免接口变更；实际同时回收两个索引，
+// 否则 user 索引的过期成员没有任何清理路径，会无界累积。）
+func (c *concurrencyCache) CleanupExpiredAccountSlotKeys(ctx context.Context) error {
+	if err := c.reconcileExpiredIndexCandidates(ctx, accountSlotIndex); err != nil {
+		return err
 	}
-	if err := c.rdb.ZRemRangeByScore(ctx, accountActiveIndexKey, "-inf", strconv.FormatInt(now, 10)).Err(); err != nil {
-		return nil, fmt.Errorf("cleanup account active index: %w", err)
-	}
-	members, err := c.rdb.ZRangeByScore(ctx, accountActiveIndexKey, &redis.ZRangeBy{
-		Min: strconv.FormatInt(now+1, 10),
-		Max: "+inf",
-	}).Result()
-	if err != nil {
-		return nil, fmt.Errorf("read account active index: %w", err)
-	}
-
-	loads, staleMembers, err := c.readAccountIndexLoads(ctx, members, now)
-	if err != nil {
-		return nil, err
-	}
-
-	loadMap := make(map[int64]*service.AccountLoadInfo, len(loads))
-	for _, load := range loads {
-		if load.slotCount == 0 && load.waitCount <= 0 {
-			// 索引候选已没有实际负载，删除 member 而不是返回空负载。
-			staleMembers = append(staleMembers, load.member)
-			continue
-		}
-		loadMap[load.id] = &service.AccountLoadInfo{
-			AccountID:          load.id,
-			CurrentConcurrency: load.slotCount,
-			WaitingCount:       load.waitCount,
-		}
-		c.touchActiveIndexForLoad(ctx, accountActiveIndexKey, load)
-	}
-	c.removeActiveIndexMembers(ctx, accountActiveIndexKey, staleMembers)
-	return loadMap, nil
+	return c.reconcileExpiredIndexCandidates(ctx, userSlotIndex)
 }
 
-// CleanupExpiredAccountSlotKeys 只处理索引中过期的账号候选。
-// 若候选仍有真实负载，则刷新索引；若没有负载，则移除索引 member。
-func (c *concurrencyCache) CleanupExpiredAccountSlotKeys(ctx context.Context) error {
+// reconcileExpiredIndexCandidates 处理单个活跃索引中 score 已到期的候选：
+// 无真实负载则移除 member；仍有负载则按真实负载批量刷新 score。
+func (c *concurrencyCache) reconcileExpiredIndexCandidates(ctx context.Context, spec slotIndexSpec) error {
 	now, err := c.redisUnixSeconds(ctx)
 	if err != nil {
 		return err
 	}
-	members, err := c.rdb.ZRangeByScore(ctx, accountActiveIndexKey, &redis.ZRangeBy{
+	members, err := c.rdb.ZRangeByScore(ctx, spec.indexKey, &redis.ZRangeBy{
 		Min:   "-inf",
 		Max:   strconv.FormatInt(now, 10),
 		Count: activeIndexCleanupBatchSize,
 	}).Result()
 	if err != nil {
-		return fmt.Errorf("read expired account active index: %w", err)
+		return fmt.Errorf("read expired index %s: %w", spec.indexKey, err)
 	}
 
-	loads, staleMembers, err := c.readAccountIndexLoads(ctx, members, now)
+	loads, staleMembers, err := c.readIndexLoads(ctx, spec, members, now)
 	if err != nil {
 		return err
 	}
+	refreshed := make([]redis.Z, 0, len(loads))
 	for _, load := range loads {
 		if load.slotCount == 0 && load.waitCount <= 0 {
 			// 真实槽位和等待数都为空，说明这个索引 member 已经完成使命。
 			staleMembers = append(staleMembers, load.member)
 			continue
 		}
-		c.touchActiveIndexForLoad(ctx, accountActiveIndexKey, load)
+		refreshed = append(refreshed, redis.Z{
+			Score:  float64(now + int64(c.activeIndexTTL(load.slotCount, load.waitCount))),
+			Member: load.member,
+		})
 	}
-	c.removeActiveIndexMembers(ctx, accountActiveIndexKey, staleMembers)
+	if len(refreshed) > 0 {
+		if err := c.rdb.ZAdd(ctx, spec.indexKey, refreshed...).Err(); err != nil {
+			logger.LegacyPrintf("repository.concurrency", "Warning: refresh %d active index members in %s failed: %v", len(refreshed), spec.indexKey, err)
+		}
+	}
+	c.removeActiveIndexMembers(ctx, spec.indexKey, staleMembers)
 	return nil
 }
 
 // CleanupStaleProcessSlots 启动时清理非当前进程前缀的槽位。
-// 清理范围来自活跃索引，避免在 Redis 上 SCAN 全部 concurrency:* 键。
+// 清理范围来自活跃索引（含 score 已过期的成员——它们往往正是崩溃进程留下的残留），
+// 避免在 Redis 上 SCAN 全部 concurrency:* 键；另有一次性迁移清扫兜底索引机制上线前的遗留等待计数。
 // API Key 槽位（concurrency:api_key:*）是 stats-only 数据：每次 Track/读取都会按分数
 // 裁剪过期成员，key 自带 TTL，可在一个 slot TTL 内自愈，因此不参与启动清理。
 func (c *concurrencyCache) CleanupStaleProcessSlots(ctx context.Context, activeRequestPrefix string) error {
 	if activeRequestPrefix == "" {
 		return nil
 	}
+	if err := c.sweepLegacyWaitKeysOnce(ctx); err != nil {
+		return err
+	}
 	now, err := c.redisUnixSeconds(ctx)
 	if err != nil {
 		return err
 	}
 
-	accountMembers, err := c.activeIndexMembers(ctx, accountActiveIndexKey, now)
+	accountMembers, err := c.allIndexMembers(ctx, accountActiveIndexKey)
 	if err != nil {
 		return err
 	}
-	if err := c.cleanupStaleProcessSlotsForIndex(ctx, accountActiveIndexKey, accountMembers, activeRequestPrefix, accountSlotKey, accountWaitKey, c.refreshAccountActiveIndex); err != nil {
+	if err := c.cleanupStaleProcessSlotsForIndex(ctx, accountSlotIndex, accountMembers, activeRequestPrefix, now); err != nil {
 		return err
 	}
 
-	userMembers, err := c.activeIndexMembers(ctx, userActiveIndexKey, now)
+	userMembers, err := c.allIndexMembers(ctx, userActiveIndexKey)
 	if err != nil {
 		return err
 	}
-	return c.cleanupStaleProcessSlotsForIndex(ctx, userActiveIndexKey, userMembers, activeRequestPrefix, userSlotKey, waitQueueKey, c.refreshUserActiveIndex)
+	return c.cleanupStaleProcessSlotsForIndex(ctx, userSlotIndex, userMembers, activeRequestPrefix, now)
 }
 
-// activeIndexMembers 只返回当前仍未过期的索引 member；过期 member 由对应清理任务处理。
-func (c *concurrencyCache) activeIndexMembers(ctx context.Context, indexKey string, now int64) ([]string, error) {
-	members, err := c.rdb.ZRangeByScore(ctx, indexKey, &redis.ZRangeBy{
-		Min: strconv.FormatInt(now+1, 10),
-		Max: "+inf",
-	}).Result()
+// sweepLegacyWaitKeysOnce 一次性清扫活跃索引机制上线前遗留的等待计数键。
+// 等待计数在有流量时会不断刷新 TTL、无法自然过期，而索引不认识旧键，
+// 因此这里例外地做一次 SCAN，用 marker 键保证整个 Redis 数据生命周期内只执行一次。
+// 先清扫后写 marker：清扫失败时下次启动会重试；并发实例重复清扫是幂等的。
+func (c *concurrencyCache) sweepLegacyWaitKeysOnce(ctx context.Context) error {
+	exists, err := c.rdb.Exists(ctx, legacyWaitSweepMarkerKey).Result()
+	if err != nil {
+		return fmt.Errorf("check legacy wait sweep marker: %w", err)
+	}
+	if exists > 0 {
+		return nil
+	}
+	for _, pattern := range []string{accountWaitKeyPrefix + "*", waitQueueKeyPrefix + "*"} {
+		var cursor uint64
+		for {
+			keys, next, err := c.rdb.Scan(ctx, cursor, pattern, 200).Result()
+			if err != nil {
+				return fmt.Errorf("scan legacy wait keys %s: %w", pattern, err)
+			}
+			if len(keys) > 0 {
+				if err := c.rdb.Del(ctx, keys...).Err(); err != nil {
+					return fmt.Errorf("delete legacy wait keys: %w", err)
+				}
+			}
+			cursor = next
+			if cursor == 0 {
+				break
+			}
+		}
+	}
+	if err := c.rdb.Set(ctx, legacyWaitSweepMarkerKey, "1", 0).Err(); err != nil {
+		return fmt.Errorf("set legacy wait sweep marker: %w", err)
+	}
+	return nil
+}
+
+// allIndexMembers 返回索引中全部 member（含 score 已过期的）。
+// 启动清理必须覆盖过期成员：长时间停机后 score 过期的候选恰恰最可能持有死进程残留。
+func (c *concurrencyCache) allIndexMembers(ctx context.Context, indexKey string) ([]string, error) {
+	members, err := c.rdb.ZRange(ctx, indexKey, 0, -1).Result()
 	if err != nil {
 		return nil, fmt.Errorf("read active index %s: %w", indexKey, err)
 	}
@@ -929,17 +983,17 @@ func (c *concurrencyCache) activeIndexMembers(ctx context.Context, indexKey stri
 }
 
 // cleanupStaleProcessSlotsForIndex 逐个处理索引中的账号/用户。
-// Lua 脚本一次只碰一个槽位 key，兼容 Redis Cluster，随后删除重启后已失效的等待计数。
+// Lua 脚本一次只碰一个槽位 key，兼容 Redis Cluster，随后删除重启后已失效的等待计数；
+// 索引 member 的去留由脚本返回的剩余槽位数决定，最后批量写回。
 func (c *concurrencyCache) cleanupStaleProcessSlotsForIndex(
 	ctx context.Context,
-	indexKey string,
+	spec slotIndexSpec,
 	members []string,
 	activeRequestPrefix string,
-	slotKeyForID func(int64) string,
-	waitKeyForID func(int64) string,
-	refreshIndex func(context.Context, int64),
+	now int64,
 ) error {
 	staleMembers := make([]string, 0)
+	refreshed := make([]redis.Z, 0)
 	for _, member := range members {
 		id, err := strconv.ParseInt(member, 10, 64)
 		if err != nil || id <= 0 {
@@ -947,14 +1001,28 @@ func (c *concurrencyCache) cleanupStaleProcessSlotsForIndex(
 			continue
 		}
 
-		if _, err := startupCleanupSlotScript.Run(ctx, c.rdb, []string{slotKeyForID(id)}, activeRequestPrefix, c.slotTTLSeconds).Result(); err != nil {
-			return fmt.Errorf("cleanup stale process slots %s: %w", slotKeyForID(id), err)
+		_, remaining, err := runScriptInt64Pair(ctx, c.rdb, startupCleanupSlotScript, []string{spec.slotKey(id)}, activeRequestPrefix, c.slotTTLSeconds)
+		if err != nil {
+			return fmt.Errorf("cleanup stale process slots %s: %w", spec.slotKey(id), err)
 		}
-		if err := c.rdb.Del(ctx, waitKeyForID(id)).Err(); err != nil {
-			return fmt.Errorf("delete stale wait key %s: %w", waitKeyForID(id), err)
+		// 等待计数属于已死进程，直接删除；剩余槽位（当前进程前缀）决定索引 member 去留。
+		if err := c.rdb.Del(ctx, spec.waitKey(id)).Err(); err != nil {
+			return fmt.Errorf("delete stale wait key %s: %w", spec.waitKey(id), err)
+		}
+		if remaining > 0 {
+			refreshed = append(refreshed, redis.Z{
+				Score:  float64(now + int64(c.slotTTLSeconds)),
+				Member: member,
+			})
+		} else {
+			staleMembers = append(staleMembers, member)
 		}
-		refreshIndex(ctx, id)
 	}
-	c.removeActiveIndexMembers(ctx, indexKey, staleMembers)
+	if len(refreshed) > 0 {
+		if err := c.rdb.ZAdd(ctx, spec.indexKey, refreshed...).Err(); err != nil {
+			logger.LegacyPrintf("repository.concurrency", "Warning: refresh %d active index members in %s failed: %v", len(refreshed), spec.indexKey, err)
+		}
+	}
+	c.removeActiveIndexMembers(ctx, spec.indexKey, staleMembers)
 	return nil
 }

backend/internal/repository/concurrency_cache_integration_test.go

@@ -77,59 +77,61 @@ func (s *ConcurrencyCacheSuite) TestAccountSlot_AcquireAndRelease() {
 	require.Equal(s.T(), 1, cur, "expected 1 after release")
 }
 
-func (s *ConcurrencyCacheSuite) TestActiveAccountLoadMap_AcquireAndRelease() {
+func (s *ConcurrencyCacheSuite) TestAccountActiveIndex_AcquireAndRelease() {
 	accountID := int64(610)
-	reqID := "active-load-req"
+	member := strconv.FormatInt(accountID, 10)
+	reqID := "active-index-req"
+
+	now, err := s.rawCache.redisUnixSeconds(s.ctx)
+	require.NoError(s.T(), err)
 
 	ok, err := s.cache.AcquireAccountSlot(s.ctx, accountID, 2, reqID)
 	require.NoError(s.T(), err)
 	require.True(s.T(), ok)
 
-	loadMap, err := s.rawCache.GetActiveAccountLoadMap(s.ctx)
+	score, err := s.rdb.ZScore(s.ctx, accountActiveIndexKey, member).Result()
 	require.NoError(s.T(), err)
-	require.Contains(s.T(), loadMap, accountID)
-	require.Equal(s.T(), 1, loadMap[accountID].CurrentConcurrency)
+	require.Greater(s.T(), int64(score), now, "index score should be a future expiry")
 
 	require.NoError(s.T(), s.cache.ReleaseAccountSlot(s.ctx, accountID, reqID))
 
-	loadMap, err = s.rawCache.GetActiveAccountLoadMap(s.ctx)
-	require.NoError(s.T(), err)
-	require.NotContains(s.T(), loadMap, accountID)
+	_, err = s.rdb.ZScore(s.ctx, accountActiveIndexKey, member).Result()
+	require.ErrorIs(s.T(), err, redis.Nil, "index member should be removed after load drops to zero")
 }
 
-func (s *ConcurrencyCacheSuite) TestActiveAccountLoadMap_AccountWaitIndexLifecycle() {
+func (s *ConcurrencyCacheSuite) TestAccountActiveIndex_WaitLifecycle() {
 	accountID := int64(611)
+	member := strconv.FormatInt(accountID, 10)
 
 	ok, err := s.cache.IncrementAccountWaitCount(s.ctx, accountID, 2)
 	require.NoError(s.T(), err)
 	require.True(s.T(), ok)
 
-	loadMap, err := s.rawCache.GetActiveAccountLoadMap(s.ctx)
-	require.NoError(s.T(), err)
-	require.Contains(s.T(), loadMap, accountID)
-	require.Equal(s.T(), 1, loadMap[accountID].WaitingCount)
+	_, err = s.rdb.ZScore(s.ctx, accountActiveIndexKey, member).Result()
+	require.NoError(s.T(), err, "wait increment should register index member")
 
 	require.NoError(s.T(), s.cache.DecrementAccountWaitCount(s.ctx, accountID))
 
-	loadMap, err = s.rawCache.GetActiveAccountLoadMap(s.ctx)
-	require.NoError(s.T(), err)
-	require.NotContains(s.T(), loadMap, accountID)
+	_, err = s.rdb.ZScore(s.ctx, accountActiveIndexKey, member).Result()
+	require.ErrorIs(s.T(), err, redis.Nil, "index member should be removed after wait drops to zero")
 }
 
-func (s *ConcurrencyCacheSuite) TestActiveAccountLoadMap_RemovesInvalidIndexMember() {
-	now, err := s.rawCache.redisUnixSeconds(s.ctx)
-	require.NoError(s.T(), err)
-	require.NoError(s.T(), s.rdb.ZAdd(s.ctx, accountActiveIndexKey, redis.Z{
-		Score:  float64(now + 60),
-		Member: "not-an-account-id",
-	}).Err())
+func (s *ConcurrencyCacheSuite) TestUserActiveIndex_AcquireAndRelease() {
+	userID := int64(612)
+	member := strconv.FormatInt(userID, 10)
+	reqID := "user-active-index-req"
 
-	loadMap, err := s.rawCache.GetActiveAccountLoadMap(s.ctx)
+	ok, err := s.cache.AcquireUserSlot(s.ctx, userID, 2, reqID)
 	require.NoError(s.T(), err)
-	require.Empty(s.T(), loadMap)
+	require.True(s.T(), ok)
 
-	_, err = s.rdb.ZScore(s.ctx, accountActiveIndexKey, "not-an-account-id").Result()
-	require.ErrorIs(s.T(), err, redis.Nil)
+	_, err = s.rdb.ZScore(s.ctx, userActiveIndexKey, member).Result()
+	require.NoError(s.T(), err, "acquire should register user index member")
+
+	require.NoError(s.T(), s.cache.ReleaseUserSlot(s.ctx, userID, reqID))
+
+	_, err = s.rdb.ZScore(s.ctx, userActiveIndexKey, member).Result()
+	require.ErrorIs(s.T(), err, redis.Nil, "user index member should be removed after release")
 }
 
 func (s *ConcurrencyCacheSuite) TestAccountSlot_TTL() {
@@ -351,6 +353,8 @@ func (s *ConcurrencyCacheSuite) TestAccountWaitQueue_IncrementAndDecrement() {
 }
 
 func (s *ConcurrencyCacheSuite) TestCleanupStaleProcessSlots() {
+	// 预置迁移 marker，隔离一次性清扫，只验证索引驱动的清理路径。
+	require.NoError(s.T(), s.rdb.Set(s.ctx, legacyWaitSweepMarkerKey, "1", 0).Err())
 	accountID := int64(901)
 	userID := int64(902)
 	apiKeyID := int64(903)
@@ -619,7 +623,113 @@ func (s *ConcurrencyCacheSuite) TestCleanupExpiredAccountSlotKeys() {
 	require.ErrorIs(s.T(), err, redis.Nil)
 }
 
+func (s *ConcurrencyCacheSuite) TestCleanupExpiredAccountSlotKeys_ReapsUserIndex() {
+	now, err := s.rawCache.redisUnixSeconds(s.ctx)
+	require.NoError(s.T(), err)
+	expiredScore := float64(now - 10)
+	userKeyWithFresh := fmt.Sprintf("%s%d", userSlotKeyPrefix, 401)
+
+	// 401 有真实负载但索引 score 已过期：应刷新而不是删除。
+	require.NoError(s.T(), s.rdb.ZAdd(s.ctx, userKeyWithFresh,
+		redis.Z{Score: float64(now), Member: "fresh"},
+	).Err())
+	// 402 无任何负载：过期索引 member 应被回收。
+	// 非法 member 也应随过期候选一并清除。
+	require.NoError(s.T(), s.rdb.ZAdd(s.ctx, userActiveIndexKey,
+		redis.Z{Score: expiredScore, Member: "401"},
+		redis.Z{Score: expiredScore, Member: "402"},
+		redis.Z{Score: expiredScore, Member: "not-a-user-id"},
+	).Err())
+
+	require.NoError(s.T(), s.cache.CleanupExpiredAccountSlotKeys(s.ctx))
+
+	score, err := s.rdb.ZScore(s.ctx, userActiveIndexKey, "401").Result()
+	require.NoError(s.T(), err)
+	require.Greater(s.T(), int64(score), now, "loaded user should be re-scheduled, not dropped")
+
+	_, err = s.rdb.ZScore(s.ctx, userActiveIndexKey, "402").Result()
+	require.ErrorIs(s.T(), err, redis.Nil, "idle expired user member should be reaped")
+
+	_, err = s.rdb.ZScore(s.ctx, userActiveIndexKey, "not-a-user-id").Result()
+	require.ErrorIs(s.T(), err, redis.Nil, "invalid member should be reaped")
+}
+
+func (s *ConcurrencyCacheSuite) TestCleanupStaleProcessSlots_LegacyWaitSweepRunsOnce() {
+	unindexedAccountWaitKey := fmt.Sprintf("%s%d", accountWaitKeyPrefix, 2901)
+	unindexedUserWaitKey := fmt.Sprintf("%s%d", waitQueueKeyPrefix, 2902)
+	require.NoError(s.T(), s.rdb.Set(s.ctx, unindexedAccountWaitKey, 5, time.Minute).Err())
+	require.NoError(s.T(), s.rdb.Set(s.ctx, unindexedUserWaitKey, 3, time.Minute).Err())
+
+	// 首次运行：marker 不存在，一次性清扫删除所有遗留等待计数（含未入索引的）。
+	require.NoError(s.T(), s.cache.CleanupStaleProcessSlots(s.ctx, "keep-"))
+
+	_, err := s.rdb.Get(s.ctx, unindexedAccountWaitKey).Result()
+	require.ErrorIs(s.T(), err, redis.Nil, "legacy account wait key should be swept on first startup")
+	_, err = s.rdb.Get(s.ctx, unindexedUserWaitKey).Result()
+	require.ErrorIs(s.T(), err, redis.Nil, "legacy user wait key should be swept on first startup")
+
+	exists, err := s.rdb.Exists(s.ctx, legacyWaitSweepMarkerKey).Result()
+	require.NoError(s.T(), err)
+	require.EqualValues(s.T(), 1, exists, "sweep marker should be set after first run")
+
+	// 再次运行：marker 已存在，未入索引的等待计数不再被触碰。
+	require.NoError(s.T(), s.rdb.Set(s.ctx, unindexedAccountWaitKey, 5, time.Minute).Err())
+	require.NoError(s.T(), s.cache.CleanupStaleProcessSlots(s.ctx, "keep-"))
+	val, err := s.rdb.Get(s.ctx, unindexedAccountWaitKey).Int()
+	require.NoError(s.T(), err, "sweep must not run twice")
+	require.Equal(s.T(), 5, val)
+}
+
+func (s *ConcurrencyCacheSuite) TestCleanupStaleProcessSlots_ProcessesExpiredIndexMembers() {
+	// score 已过期的索引成员往往正是崩溃进程留下的残留，启动清理必须覆盖它们。
+	require.NoError(s.T(), s.rdb.Set(s.ctx, legacyWaitSweepMarkerKey, "1", 0).Err())
+	accountID := int64(3901)
+	userID := int64(3902)
+	accountKey := fmt.Sprintf("%s%d", accountSlotKeyPrefix, accountID)
+	userKey := fmt.Sprintf("%s%d", userSlotKeyPrefix, userID)
+	accountWaitKey := fmt.Sprintf("%s%d", accountWaitKeyPrefix, accountID)
+
+	now, err := s.rawCache.redisUnixSeconds(s.ctx)
+	require.NoError(s.T(), err)
+	require.NoError(s.T(), s.rdb.ZAdd(s.ctx, accountKey,
+		redis.Z{Score: float64(now), Member: "oldproc-1"},
+	).Err())
+	require.NoError(s.T(), s.rdb.ZAdd(s.ctx, userKey,
+		redis.Z{Score: float64(now), Member: "oldproc-2"},
+	).Err())
+	require.NoError(s.T(), s.rdb.Set(s.ctx, accountWaitKey, 4, time.Minute).Err())
+	// 索引 score 设为过去时刻，模拟长时间停机后索引已“过期”。
+	require.NoError(s.T(), s.rdb.ZAdd(s.ctx, accountActiveIndexKey, redis.Z{
+		Score:  float64(now - 100),
+		Member: strconv.FormatInt(accountID, 10),
+	}).Err())
+	require.NoError(s.T(), s.rdb.ZAdd(s.ctx, userActiveIndexKey, redis.Z{
+		Score:  float64(now - 100),
+		Member: strconv.FormatInt(userID, 10),
+	}).Err())
+
+	require.NoError(s.T(), s.cache.CleanupStaleProcessSlots(s.ctx, "keep-"))
+
+	exists, err := s.rdb.Exists(s.ctx, accountKey).Result()
+	require.NoError(s.T(), err)
+	require.EqualValues(s.T(), 0, exists, "stale slot key of expired index member should be purged")
+
+	exists, err = s.rdb.Exists(s.ctx, userKey).Result()
+	require.NoError(s.T(), err)
+	require.EqualValues(s.T(), 0, exists)
+
+	_, err = s.rdb.Get(s.ctx, accountWaitKey).Result()
+	require.ErrorIs(s.T(), err, redis.Nil, "wait counter of expired index member should be deleted")
+
+	_, err = s.rdb.ZScore(s.ctx, accountActiveIndexKey, strconv.FormatInt(accountID, 10)).Result()
+	require.ErrorIs(s.T(), err, redis.Nil, "emptied member should be removed from index")
+	_, err = s.rdb.ZScore(s.ctx, userActiveIndexKey, strconv.FormatInt(userID, 10)).Result()
+	require.ErrorIs(s.T(), err, redis.Nil)
+}
+
 func (s *ConcurrencyCacheSuite) TestCleanupStaleProcessSlots_RemovesOldPrefixesAndWaitCounters() {
+	// 预置迁移 marker，确保等待计数删除来自索引驱动路径而非一次性清扫。
+	require.NoError(s.T(), s.rdb.Set(s.ctx, legacyWaitSweepMarkerKey, "1", 0).Err())
 	accountID := int64(901)
 	userID := int64(902)
 	accountSlotKey := fmt.Sprintf("%s%d", accountSlotKeyPrefix, accountID)

backend/internal/repository/user_msg_queue_cache.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"strconv"
 
+	"github.com/Wei-Shaw/sub2api/internal/pkg/logger"
 	"github.com/Wei-Shaw/sub2api/internal/service"
 	"github.com/redis/go-redis/v9"
 )
@@ -25,6 +26,9 @@ const (
 
 // Lua 脚本：原子获取串行锁（SET NX PX + 重入安全）
 // 返回 {是否获取成功, 锁预计过期时间毫秒}，让 Go 侧用同一 Redis 时间源更新索引。
+// 获取失败（锁被他人持有）时也返回观测到的到期时间，供 Go 侧回填锁索引：
+// 这让升级窗口遗留、索引写失败、释放竞态误删索引的存量锁在下一次被争用时自动重新入索引，
+// 是替代旧 SCAN 兜底的自愈机制。PTTL == -1 的异常锁返回当前时间，使其立即成为 reconcile 候选。
 var acquireLockScript = redis.NewScript(`
 redis.replicate_commands()
 local cur = redis.call('GET', KEYS[1])
@@ -35,7 +39,15 @@ if cur == ARGV[1] then
     local ms = tonumber(t[1])*1000 + math.floor(tonumber(t[2])/1000)
     return {1, ms + ttl}
 end
-if cur ~= false then return {0, 0} end
+if cur ~= false then
+    local t = redis.call('TIME')
+    local ms = tonumber(t[1])*1000 + math.floor(tonumber(t[2])/1000)
+    local pttl = redis.call('PTTL', KEYS[1])
+    if pttl and pttl > 0 then
+        return {0, ms + pttl}
+    end
+    return {0, ms}
+end
 redis.call('SET', KEYS[1], ARGV[1], 'PX', ttl)
 local t = redis.call('TIME')
 local ms = tonumber(t[1])*1000 + math.floor(tonumber(t[2])/1000)
@@ -92,7 +104,8 @@ func umqLastKey(accountID int64) string {
 }
 
 // AcquireLock 尝试获取账号级串行锁
-// 成功后尽力写入锁索引，后台清理只需要看“到期候选”而不是扫描所有锁 key。
+// 无论成功与否都尽力写入锁索引：成功时登记自己的锁，失败时回填观测到的持有者锁，
+// 保证任何被争用的锁都能被后台 reconcile 发现，无需扫描所有锁 key。
 func (c *userMsgQueueCache) AcquireLock(ctx context.Context, accountID int64, requestID string, lockTtlMs int) (bool, error) {
 	key := umqLockKey(accountID)
 	result, err := acquireLockScript.Run(ctx, c.rdb, []string{key}, requestID, lockTtlMs).Result()
@@ -107,11 +120,13 @@ func (c *userMsgQueueCache) AcquireLock(ctx context.Context, accountID int64, re
 	if err != nil {
 		return false, fmt.Errorf("umq parse acquire lock expire: %w", err)
 	}
-	if acquired == 1 {
-		_ = c.rdb.ZAdd(ctx, umqLockIndexKey, redis.Z{
+	if expireAtMs > 0 {
+		if err := c.rdb.ZAdd(ctx, umqLockIndexKey, redis.Z{
 			Score:  float64(expireAtMs),
 			Member: strconv.FormatInt(accountID, 10),
-		}).Err()
+		}).Err(); err != nil {
+			logger.LegacyPrintf("repository.umq", "Warning: update lock index for account %d failed: %v", accountID, err)
+		}
 	}
 	return acquired == 1, nil
 }
@@ -126,7 +141,11 @@ func (c *userMsgQueueCache) ReleaseLock(ctx context.Context, accountID int64, re
 		return false, fmt.Errorf("umq release lock: %w", err)
 	}
 	if result == 1 {
-		_ = c.rdb.ZRem(ctx, umqLockIndexKey, strconv.FormatInt(accountID, 10)).Err()
+		// 与下一个 AcquireLock 的 ZAdd 存在竞态：可能误删新持有者刚写入的索引项。
+		// 该锁下次被争用时 AcquireLock 的回填路径会重新登记，无需在此加锁。
+		if err := c.rdb.ZRem(ctx, umqLockIndexKey, strconv.FormatInt(accountID, 10)).Err(); err != nil {
+			logger.LegacyPrintf("repository.umq", "Warning: remove lock index for account %d failed: %v", accountID, err)
+		}
 	}
 	return result == 1, nil
 }
@@ -180,7 +199,7 @@ func (c *userMsgQueueCache) ReconcileExpiredLockCandidates(ctx context.Context,
 	for _, member := range members {
 		accountID, err := strconv.ParseInt(member, 10, 64)
 		if err != nil || accountID <= 0 {
-			_ = c.rdb.ZRem(ctx, umqLockIndexKey, member).Err()
+			c.removeLockIndexMember(ctx, member)
 			continue
 		}
 
@@ -200,22 +219,31 @@ func (c *userMsgQueueCache) ReconcileExpiredLockCandidates(ctx context.Context,
 		switch status {
 		case -2:
 			// 锁自然过期或已释放，只需移除索引残留。
-			_ = c.rdb.ZRem(ctx, umqLockIndexKey, member).Err()
+			c.removeLockIndexMember(ctx, member)
 		case -1:
 			// 无 TTL 的锁会永久阻塞队列，Lua 已原子删除它，这里统计一次清理。
-			_ = c.rdb.ZRem(ctx, umqLockIndexKey, member).Err()
+			c.removeLockIndexMember(ctx, member)
 			cleaned++
 		case 1:
 			// 锁仍存活，说明索引过期时间滞后；按剩余 PTTL 重新排期。
-			_ = c.rdb.ZAdd(ctx, umqLockIndexKey, redis.Z{
+			if err := c.rdb.ZAdd(ctx, umqLockIndexKey, redis.Z{
 				Score:  float64(nowMs + pttl),
 				Member: member,
-			}).Err()
+			}).Err(); err != nil {
+				logger.LegacyPrintf("repository.umq", "Warning: reschedule lock index member %s failed: %v", member, err)
+			}
 		}
 	}
 	return cleaned, nil
 }
 
+// removeLockIndexMember 移除锁索引残留；索引维护是 best-effort，失败只记日志。
+func (c *userMsgQueueCache) removeLockIndexMember(ctx context.Context, member string) {
+	if err := c.rdb.ZRem(ctx, umqLockIndexKey, member).Err(); err != nil {
+		logger.LegacyPrintf("repository.umq", "Warning: remove lock index member %s failed: %v", member, err)
+	}
+}
+
 // redisScriptInt64At 兼容 go-redis 对 Lua 数组元素的不同返回类型。
 func redisScriptInt64At(result any, index int) (int64, error) {
 	values, ok := result.([]any)

backend/internal/repository/user_msg_queue_cache_integration_test.go

@@ -126,3 +126,52 @@ func (s *UserMsgQueueCacheSuite) TestReconcileExpiredLockCandidatesRemovesInvali
 	_, err = s.rdb.ZScore(s.ctx, umqLockIndexKey, "not-an-account-id").Result()
 	require.True(s.T(), errors.Is(err, redis.Nil))
 }
+
+func (s *UserMsgQueueCacheSuite) TestAcquireLockBusyPathReindexesUnindexedLiveLock() {
+	// 模拟索引丢失的存量锁（升级窗口/索引写失败/释放竞态误删）：
+	// 锁存在且有 TTL，但索引里没有对应 member。
+	accountID := int64(705)
+	nowMs, err := s.cache.GetCurrentTimeMs(s.ctx)
+	require.NoError(s.T(), err)
+	require.NoError(s.T(), s.rdb.Set(s.ctx, umqLockKey(accountID), "holder-705", time.Minute).Err())
+
+	// 另一个请求争锁失败，应顺手把观测到的持有者锁回填进索引。
+	acquired, err := s.cache.AcquireLock(s.ctx, accountID, "contender-705", 10_000)
+	require.NoError(s.T(), err)
+	require.False(s.T(), acquired)
+
+	score, err := s.rdb.ZScore(s.ctx, umqLockIndexKey, "705").Result()
+	require.NoError(s.T(), err, "busy acquire should re-index the observed live lock")
+	require.Greater(s.T(), int64(score), nowMs)
+	// 锁本身不应被争锁方改动。
+	val, err := s.rdb.Get(s.ctx, umqLockKey(accountID)).Result()
+	require.NoError(s.T(), err)
+	require.Equal(s.T(), "holder-705", val)
+}
+
+func (s *UserMsgQueueCacheSuite) TestAcquireLockBusyPathMakesNoTTLLockReconcilable() {
+	// PTTL == -1 的异常锁若不在索引中，永远不会被 reconcile 发现；
+	// 争锁失败路径必须以“已到期候选”的 score 回填它，形成自愈闭环。
+	accountID := int64(706)
+	require.NoError(s.T(), s.rdb.Set(s.ctx, umqLockKey(accountID), "holder-706", 0).Err())
+
+	acquired, err := s.cache.AcquireLock(s.ctx, accountID, "contender-706", 10_000)
+	require.NoError(s.T(), err)
+	require.False(s.T(), acquired)
+
+	nowMs, err := s.cache.GetCurrentTimeMs(s.ctx)
+	require.NoError(s.T(), err)
+	score, err := s.rdb.ZScore(s.ctx, umqLockIndexKey, "706").Result()
+	require.NoError(s.T(), err, "busy acquire should index the anomalous lock")
+	require.LessOrEqual(s.T(), int64(score), nowMs, "anomalous lock should be an immediately-expired candidate")
+
+	cleaned, err := s.cache.ReconcileExpiredLockCandidates(s.ctx, 1000)
+	require.NoError(s.T(), err)
+	require.Equal(s.T(), 1, cleaned, "reconcile should delete the no-TTL lock")
+
+	exists, err := s.rdb.Exists(s.ctx, umqLockKey(accountID)).Result()
+	require.NoError(s.T(), err)
+	require.EqualValues(s.T(), 0, exists, "queue is unblocked after reconcile")
+	_, err = s.rdb.ZScore(s.ctx, umqLockIndexKey, "706").Result()
+	require.ErrorIs(s.T(), err, redis.Nil)
+}