opensource/sub2api
3
0
Fork 0
mirror of https://github.com/Wei-Shaw/sub2api.git synced 2026-09-24 16:05:44 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(redis-index): harden index-driven cleanup introduced in #3762

Browse Source 
- 启动清理覆盖 score 已过期的索引成员，并新增 marker 门控的一次性
  遗留 wait 键清扫，消除重启后幻影等待计数挤占 maxWait 的问题
- UMQ 争锁失败路径回填观测到的持有者锁索引（PTTL=-1 异常锁以已到期
  候选入索引），恢复被 SCAN 删除后丢失的孤儿锁兜底完整性
- 后台清理同时回收 account 与 user 两个活跃索引的到期候选，修复
  user 索引过期成员无回收路径导致的无界增长
- acquire/incrementWait Lua 脚本返回 {result, now}，索引写复用脚本
  内 Redis 时间，热路径每次操作减少一次 TIME 往返；清理循环外提
  now 并批量写索引
- 删除无生产调用方的 GetActiveAccountLoadMap 及配套死代码
- 索引维护失败增加 warn 日志，不再静默丢弃
- 移除根目录工单文档 REDIS_SCAN_ARCHITECTURE_OPTIMIZATION.md
This commit is contained in:
shaw
2026-07-07 10:04:38 +08:00
parent db9d92ba8f
commit 7650cce5a4
5 changed files with 424 additions and 737 deletions
Download Patch File Download Diff File Expand all files Collapse all files
REDIS_SCAN_ARCHITECTURE_OPTIMIZATION.md
-568
View File
@@ -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
+199 -131
View File
@@ -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
+137 -27
View File
@@ -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
+39 -11
View File
@@ -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
+49
View File
@@ -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)
}