feat(queue): 优化用户消息队列锁释放时机

将队列锁释放时机从"请求完成后"提前到"请求发送后",因为 Claude API
限流(RPM)基于请求发送时刻计算,无需等待响应完成。

主要变更:
- 移除锁续租机制(startLockRenewal、refreshUserMessageLock)
- 所有 relay 服务在请求发送成功后立即释放锁
- 流式请求通过 onResponseStart 回调在收到响应头时释放
- 调整默认配置:timeoutMs 60s→5s,lockTtlMs 120s→5s
- 新增 USER_MESSAGE_QUEUE_LOCK_TTL_MS 环境变量支持
This commit is contained in:
QTom
2025-12-10 01:26:00 +08:00
committed by QTom
parent cb94a4260e
commit 3b9c96dff8
11 changed files with 251 additions and 314 deletions

View File

@@ -179,88 +179,6 @@ describe('UserMessageQueueService', () => {
})
})
describe('startLockRenewal', () => {
beforeEach(() => {
jest.useFakeTimers()
})
afterEach(() => {
jest.useRealTimers()
jest.restoreAllMocks()
})
it('should periodically refresh lock while enabled', async () => {
jest.spyOn(userMessageQueueService, 'getConfig').mockResolvedValue({
enabled: true,
delayMs: 200,
timeoutMs: 30000,
lockTtlMs: 120000
})
const refreshSpy = jest.spyOn(redis, 'refreshUserMessageLock').mockResolvedValue(true)
const stop = await userMessageQueueService.startLockRenewal('acct-1', 'req-1')
jest.advanceTimersByTime(60000) // 半个TTL
await Promise.resolve()
expect(refreshSpy).toHaveBeenCalledWith('acct-1', 'req-1', 120000)
stop()
})
it('should no-op when queue disabled', async () => {
jest.spyOn(userMessageQueueService, 'getConfig').mockResolvedValue({
enabled: false,
delayMs: 200,
timeoutMs: 30000,
lockTtlMs: 120000
})
const refreshSpy = jest.spyOn(redis, 'refreshUserMessageLock').mockResolvedValue(true)
const stop = await userMessageQueueService.startLockRenewal('acct-1', 'req-1')
jest.advanceTimersByTime(120000)
await Promise.resolve()
expect(refreshSpy).not.toHaveBeenCalled()
stop()
})
it('should track active renewal timer', async () => {
jest.spyOn(userMessageQueueService, 'getConfig').mockResolvedValue({
enabled: true,
delayMs: 200,
timeoutMs: 30000,
lockTtlMs: 120000
})
jest.spyOn(redis, 'refreshUserMessageLock').mockResolvedValue(true)
expect(userMessageQueueService.getActiveRenewalCount()).toBe(0)
const stop = await userMessageQueueService.startLockRenewal('acct-1', 'req-1')
expect(userMessageQueueService.getActiveRenewalCount()).toBe(1)
stop()
expect(userMessageQueueService.getActiveRenewalCount()).toBe(0)
})
it('should stop all renewal timers on service shutdown', async () => {
jest.spyOn(userMessageQueueService, 'getConfig').mockResolvedValue({
enabled: true,
delayMs: 200,
timeoutMs: 30000,
lockTtlMs: 120000
})
jest.spyOn(redis, 'refreshUserMessageLock').mockResolvedValue(true)
await userMessageQueueService.startLockRenewal('acct-1', 'req-1')
await userMessageQueueService.startLockRenewal('acct-2', 'req-2')
expect(userMessageQueueService.getActiveRenewalCount()).toBe(2)
userMessageQueueService.stopAllRenewalTimers()
expect(userMessageQueueService.getActiveRenewalCount()).toBe(0)
})
})
describe('acquireQueueLock', () => {
afterEach(() => {
jest.restoreAllMocks()