diff --git a/apps/sim/app/api/cron/renew-subscriptions/route.test.ts b/apps/sim/app/api/cron/renew-subscriptions/route.test.ts index 86579f7f037..de6308d4d4d 100644 --- a/apps/sim/app/api/cron/renew-subscriptions/route.test.ts +++ b/apps/sim/app/api/cron/renew-subscriptions/route.test.ts @@ -63,7 +63,8 @@ describe('Teams subscription renewal route (fire-and-forget)', () => { expect(redisConfigMockFns.mockAcquireLock).toHaveBeenCalledWith( 'teams-subscription-renewal-lock', expect.any(String), - expect.any(Number) + expect.any(Number), + { reclaimOnFailure: true } ) await flushMicrotasks() diff --git a/apps/sim/app/api/cron/renew-subscriptions/route.ts b/apps/sim/app/api/cron/renew-subscriptions/route.ts index 8818d5462b1..08ba269b241 100644 --- a/apps/sim/app/api/cron/renew-subscriptions/route.ts +++ b/apps/sim/app/api/cron/renew-subscriptions/route.ts @@ -252,7 +252,9 @@ export const GET = withRouteHandler(async (request: NextRequest) => { } const lockValue = generateShortId() - const locked = await acquireLock(LOCK_KEY, lockValue, LOCK_TTL_SECONDS) + const locked = await acquireLock(LOCK_KEY, lockValue, LOCK_TTL_SECONDS, { + reclaimOnFailure: true, + }) if (!locked) { return NextResponse.json( { success: true, message: 'Renewal already in progress – skipped', status: 'skip' }, diff --git a/apps/sim/app/api/resume/poll/route.ts b/apps/sim/app/api/resume/poll/route.ts index be39b767fe9..c43fb682c0b 100644 --- a/apps/sim/app/api/resume/poll/route.ts +++ b/apps/sim/app/api/resume/poll/route.ts @@ -62,7 +62,9 @@ export const GET = withRouteHandler(async (request: NextRequest) => { const authError = verifyCronAuth(request, 'Time-pause resume poll') if (authError) return authError - const lockAcquired = await acquireLock(LOCK_KEY, requestId, LOCK_TTL_SECONDS) + const lockAcquired = await acquireLock(LOCK_KEY, requestId, LOCK_TTL_SECONDS, { + reclaimOnFailure: true, + }) if (!lockAcquired) { return NextResponse.json( { success: true, message: 'Polling already in progress – skipped', requestId }, diff --git a/apps/sim/app/api/webhooks/poll/[provider]/route.test.ts b/apps/sim/app/api/webhooks/poll/[provider]/route.test.ts index 432e38bf308..e2deba46f98 100644 --- a/apps/sim/app/api/webhooks/poll/[provider]/route.test.ts +++ b/apps/sim/app/api/webhooks/poll/[provider]/route.test.ts @@ -66,10 +66,13 @@ describe('webhook polling route (fire-and-forget)', () => { expect(response.status).toBe(202) const data = await response.json() expect(data).toMatchObject({ status: 'started' }) + // `reclaimOnFailure` is what stops a timed-out acquire from leaving a lock + // no one owns, which skipped every poll until the TTL expired. expect(redisConfigMockFns.mockAcquireLock).toHaveBeenCalledWith( 'gmail-polling-lock', expect.any(String), - expect.any(Number) + expect.any(Number), + { reclaimOnFailure: true } ) await flushMicrotasks() diff --git a/apps/sim/app/api/webhooks/poll/[provider]/route.ts b/apps/sim/app/api/webhooks/poll/[provider]/route.ts index a55c1082724..8c09260dbb9 100644 --- a/apps/sim/app/api/webhooks/poll/[provider]/route.ts +++ b/apps/sim/app/api/webhooks/poll/[provider]/route.ts @@ -40,7 +40,9 @@ export const GET = withRouteHandler( const LOCK_KEY = `${provider}-polling-lock` const lockValue = requestId - const locked = await acquireLock(LOCK_KEY, lockValue, LOCK_TTL_SECONDS) + const locked = await acquireLock(LOCK_KEY, lockValue, LOCK_TTL_SECONDS, { + reclaimOnFailure: true, + }) if (!locked) { return NextResponse.json( { diff --git a/apps/sim/app/api/workspace-events/poll/route.test.ts b/apps/sim/app/api/workspace-events/poll/route.test.ts index 96d52e8b3a7..fd423dbb5cb 100644 --- a/apps/sim/app/api/workspace-events/poll/route.test.ts +++ b/apps/sim/app/api/workspace-events/poll/route.test.ts @@ -62,7 +62,8 @@ describe('workspace events polling route (fire-and-forget)', () => { expect(redisConfigMockFns.mockAcquireLock).toHaveBeenCalledWith( 'workspace-events-no-activity-poll-lock', expect.any(String), - expect.any(Number) + expect.any(Number), + { reclaimOnFailure: true } ) await flushMicrotasks() diff --git a/apps/sim/app/api/workspace-events/poll/route.ts b/apps/sim/app/api/workspace-events/poll/route.ts index 6408eb0bd64..7cde86e0d49 100644 --- a/apps/sim/app/api/workspace-events/poll/route.ts +++ b/apps/sim/app/api/workspace-events/poll/route.ts @@ -31,7 +31,9 @@ export const GET = withRouteHandler(async (request: NextRequest) => { return authError } - const lockAcquired = await acquireLock(LOCK_KEY, requestId, LOCK_TTL_SECONDS) + const lockAcquired = await acquireLock(LOCK_KEY, requestId, LOCK_TTL_SECONDS, { + reclaimOnFailure: true, + }) if (!lockAcquired) { return NextResponse.json( diff --git a/apps/sim/lib/core/config/redis.test.ts b/apps/sim/lib/core/config/redis.test.ts index 69662e173b0..96b0c49e4b6 100644 --- a/apps/sim/lib/core/config/redis.test.ts +++ b/apps/sim/lib/core/config/redis.test.ts @@ -25,6 +25,7 @@ vi.mock('ioredis', () => ({ })) import { + acquireLock, closeRedisConnection, extendLock, getRedisClient, @@ -208,6 +209,72 @@ describe('redis config', () => { }) }) + describe('acquireLock', () => { + const lockKey = 'outlook-polling-lock' + const value = 'req-abc' + const ttlSeconds = 180 + + it('returns true when SET NX takes the lock', async () => { + mockRedisInstance.set.mockResolvedValueOnce('OK') + + expect(await acquireLock(lockKey, value, ttlSeconds)).toBe(true) + expect(mockRedisInstance.set).toHaveBeenCalledWith(lockKey, value, 'EX', ttlSeconds, 'NX') + expect(mockRedisInstance.eval).not.toHaveBeenCalled() + }) + + it('returns false without cleanup when the lock is already held', async () => { + mockRedisInstance.set.mockResolvedValueOnce(null) + + expect(await acquireLock(lockKey, value, ttlSeconds)).toBe(false) + expect(mockRedisInstance.eval).not.toHaveBeenCalled() + }) + + it('reclaims the lock it may have taken when SET times out and reclaim is on', async () => { + // ioredis gives up client-side on `commandTimeout` while the command can + // still land, so the lock would otherwise be held by a caller that never + // learned it won and never releases it. + mockRedisInstance.set.mockRejectedValueOnce(new Error('Command timed out')) + mockRedisInstance.eval.mockResolvedValueOnce(1) + + await expect( + acquireLock(lockKey, value, ttlSeconds, { reclaimOnFailure: true }) + ).rejects.toThrow('Command timed out') + expect(mockRedisInstance.eval).toHaveBeenCalledWith( + expect.stringContaining('del'), + 1, + lockKey, + value + ) + }) + + it('leaves the lock alone by default so a fall-open caller keeps holding it', async () => { + // `withLeaderLock` and the MCP OAuth mutex run their work anyway when + // acquisition throws. Freeing the lock under them would let a second + // runner in alongside, so reclaiming has to stay opt-in. + mockRedisInstance.set.mockRejectedValueOnce(new Error('Command timed out')) + + await expect(acquireLock(lockKey, value, ttlSeconds)).rejects.toThrow('Command timed out') + expect(mockRedisInstance.eval).not.toHaveBeenCalled() + }) + + it('surfaces the original failure when the cleanup also fails', async () => { + mockRedisInstance.set.mockRejectedValueOnce(new Error('Command timed out')) + mockRedisInstance.eval.mockRejectedValueOnce(new Error('Connection is closed')) + + // The TTL stays the backstop; the caller must still see why acquiring failed. + await expect( + acquireLock(lockKey, value, ttlSeconds, { reclaimOnFailure: true }) + ).rejects.toThrow('Command timed out') + }) + + it('returns true as a no-op when the cache capability selects the database', async () => { + mockEnv.REDIS_URL = undefined + + expect(await acquireLock(lockKey, value, ttlSeconds)).toBe(true) + expect(mockRedisInstance.set).not.toHaveBeenCalled() + }) + }) + describe('capability validation', () => { it('rejects a non-Redis URL before constructing a client', () => { mockEnv.REDIS_URL = 'https://cache.example.com' diff --git a/apps/sim/lib/core/config/redis.ts b/apps/sim/lib/core/config/redis.ts index d5bed2d4954..04ae9ae53b2 100644 --- a/apps/sim/lib/core/config/redis.ts +++ b/apps/sim/lib/core/config/redis.ts @@ -235,18 +235,51 @@ end * single-replica deployments to function without Redis. In multi-replica * deployments without Redis, the idempotency layer prevents duplicate processing. */ +export interface AcquireLockOptions { + /** + * Release the lock this call may have taken when the SET itself rejects. + * + * A rejected SET does not mean the server declined it: `commandTimeout` gives + * up client-side while the command can still reach Redis and take the lock, + * leaving it held by a caller that never learned it won and so never releases + * it. Every contender then skips until the TTL expires. + * + * Only opt in when BOTH hold, because the reclaim is unsafe otherwise: + * + * 1. `value` is unique to this holder. A value two holders can share — the + * copilot chat lock keys on a client-supplied `userMessageId` — makes the + * compare-and-delete match a lock another holder is actively using. + * 2. A throw means the caller does no work. One that falls open and runs + * anyway (`withLeaderLock`, the MCP OAuth refresh mutex) would keep running + * while this frees its lock, admitting a second concurrent runner. + */ + reclaimOnFailure?: boolean +} + export async function acquireLock( lockKey: string, value: string, - expirySeconds: number + expirySeconds: number, + options?: AcquireLockOptions ): Promise { const redis = getRedisClient() if (!redis) { return true // No-op when Redis unavailable; idempotency layer handles duplicates } - const result = await redis.set(lockKey, value, 'EX', expirySeconds, 'NX') - return result === 'OK' + try { + const result = await redis.set(lockKey, value, 'EX', expirySeconds, 'NX') + return result === 'OK' + } catch (error) { + // Best effort, and the same compare-and-delete `releaseLock` runs on the + // success path: it deletes only while `value` still owns the key. If Redis + // is still unreachable the TTL stays the backstop, which is the behavior + // without this cleanup. + if (options?.reclaimOnFailure) { + await releaseLock(lockKey, value).catch(() => {}) + } + throw error + } } /**