diff --git a/REDIS_SCAN_ARCHITECTURE_OPTIMIZATION.md b/REDIS_SCAN_ARCHITECTURE_OPTIMIZATION.md deleted file mode 100644 index 7d95419a39..0000000000 --- a/REDIS_SCAN_ARCHITECTURE_OPTIMIZATION.md +++ /dev/null @@ -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、自然过期。 diff --git a/backend/internal/repository/concurrency_cache.go b/backend/internal/repository/concurrency_cache.go index 57b18c6903..b657c1ce8f 100644 --- a/backend/internal/repository/concurrency_cache.go +++ b/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 } diff --git a/backend/internal/repository/concurrency_cache_integration_test.go b/backend/internal/repository/concurrency_cache_integration_test.go index 3c831487de..f7e27d1118 100644 --- a/backend/internal/repository/concurrency_cache_integration_test.go +++ b/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) diff --git a/backend/internal/repository/user_msg_queue_cache.go b/backend/internal/repository/user_msg_queue_cache.go index 67e75a87cf..9b7707614c 100644 --- a/backend/internal/repository/user_msg_queue_cache.go +++ b/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) diff --git a/backend/internal/repository/user_msg_queue_cache_integration_test.go b/backend/internal/repository/user_msg_queue_cache_integration_test.go index c61b658357..e683b44aa3 100644 --- a/backend/internal/repository/user_msg_queue_cache_integration_test.go +++ b/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) +}