From f808908a067071a9d832ef48668920b51b47bd7b Mon Sep 17 00:00:00 2001 From: Vikhyath Mondreti Date: Wed, 19 Aug 2026 14:19:10 -0700 Subject: [PATCH 1/2] fix(redis): reclaim a lock a timed-out acquire may have taken MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `acquireLock` awaited `SET NX` and let a rejection propagate. But a rejected SET does not mean the server declined it: the client is configured with `commandTimeout: 5000`, and ioredis gives up locally while the command can still reach Redis and take the lock. The caller never learns it won, so it never releases — and since every caller treats a throw as "did not acquire", nothing else releases it either. Every contender then skips until the TTL expires. Staging hit this on the Outlook polling cron: `acquireLock` threw `Command timed out`, the route returned 500, and the next scheduled poll and the Lambda retry both got `Polling already in progress - skipped` against a lock whose holder had never started polling. The 180s TTL cleared it. On failure, best-effort compare-and-delete through the existing `releaseLock`. That deletes only while this token still owns the key, so a lock another holder won in the meantime is untouched, and if Redis is still unreachable the TTL stays the backstop — the behavior without this cleanup. Control flow is unchanged for all nine call sites: the original error still propagates. Co-Authored-By: Claude Opus 5 (1M context) --- apps/sim/lib/core/config/redis.test.ts | 53 ++++++++++++++++++++++++++ apps/sim/lib/core/config/redis.ts | 21 +++++++++- 2 files changed, 72 insertions(+), 2 deletions(-) diff --git a/apps/sim/lib/core/config/redis.test.ts b/apps/sim/lib/core/config/redis.test.ts index 69662e173b0..a4ad9fbc295 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,58 @@ 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', 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)).rejects.toThrow('Command timed out') + expect(mockRedisInstance.eval).toHaveBeenCalledWith( + expect.stringContaining('del'), + 1, + lockKey, + value + ) + }) + + 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)).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..e79450bd980 100644 --- a/apps/sim/lib/core/config/redis.ts +++ b/apps/sim/lib/core/config/redis.ts @@ -245,8 +245,25 @@ export async function acquireLock( 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) { + /* + * A rejected SET does not mean the server declined it. `commandTimeout` + * gives up client-side while the command may 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. + * + * Reclaiming it is safe because this is the same compare-and-delete + * `releaseLock` uses on the success path: it deletes only while `value` + * still owns the key, so a lock a different holder won in the meantime is + * left alone. Best effort — if Redis is still unreachable the TTL remains + * the backstop, which is exactly the behavior without this cleanup. + */ + await releaseLock(lockKey, value).catch(() => {}) + throw error + } } /** From 121cdd40c3cbcdfdca337e84aa8c3aadfaa2bae2 Mon Sep 17 00:00:00 2001 From: Vikhyath Mondreti Date: Wed, 19 Aug 2026 15:25:43 -0700 Subject: [PATCH 2/2] fix(redis): make the timed-out-acquire reclaim opt-in per caller MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review caught that reclaiming unconditionally is unsafe for two caller classes, both of which exist today: Callers that fall open. `withLeaderLock` and the MCP OAuth refresh mutex catch a throw from `acquireLock` and run their work uncoordinated. If the SET landed, today the lock they hold keeps everyone else out while they run. Freeing it under them admits a second concurrent runner — for OAuth refresh that means two rotations of the same token and an `invalid_grant`. Callers whose lock value is not unique. The copilot chat lock keys on `streamId`, which is the client-supplied `userMessageId`. Two sends can carry the same value, so a compare-and-delete from a contender that timed out can match — and delete — the lock the active stream is holding. Reclaiming is therefore opt-in, and the option documents both preconditions it needs: a value unique to the holder, and a caller that does no work when acquisition throws. Default behavior is byte-for-byte what it was before. Opted in are the four cron/poll callers that satisfy both — webhook polling, resume polling, workspace-events polling, and Teams subscription renewal. Each mints its value with `generateShortId()` and returns 5xx rather than proceeding when acquisition throws. Co-Authored-By: Claude Opus 5 (1M context) --- .../cron/renew-subscriptions/route.test.ts | 3 +- .../app/api/cron/renew-subscriptions/route.ts | 4 +- apps/sim/app/api/resume/poll/route.ts | 4 +- .../webhooks/poll/[provider]/route.test.ts | 5 ++- .../app/api/webhooks/poll/[provider]/route.ts | 4 +- .../api/workspace-events/poll/route.test.ts | 3 +- .../app/api/workspace-events/poll/route.ts | 4 +- apps/sim/lib/core/config/redis.test.ts | 20 +++++++-- apps/sim/lib/core/config/redis.ts | 44 +++++++++++++------ 9 files changed, 67 insertions(+), 24 deletions(-) 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 a4ad9fbc295..96b0c49e4b6 100644 --- a/apps/sim/lib/core/config/redis.test.ts +++ b/apps/sim/lib/core/config/redis.test.ts @@ -229,14 +229,16 @@ describe('redis config', () => { expect(mockRedisInstance.eval).not.toHaveBeenCalled() }) - it('reclaims the lock it may have taken when SET times out', async () => { + 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)).rejects.toThrow('Command timed out') + await expect( + acquireLock(lockKey, value, ttlSeconds, { reclaimOnFailure: true }) + ).rejects.toThrow('Command timed out') expect(mockRedisInstance.eval).toHaveBeenCalledWith( expect.stringContaining('del'), 1, @@ -245,12 +247,24 @@ describe('redis config', () => { ) }) + 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)).rejects.toThrow('Command timed out') + 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 () => { diff --git a/apps/sim/lib/core/config/redis.ts b/apps/sim/lib/core/config/redis.ts index e79450bd980..04ae9ae53b2 100644 --- a/apps/sim/lib/core/config/redis.ts +++ b/apps/sim/lib/core/config/redis.ts @@ -235,10 +235,32 @@ 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) { @@ -249,19 +271,13 @@ export async function acquireLock( const result = await redis.set(lockKey, value, 'EX', expirySeconds, 'NX') return result === 'OK' } catch (error) { - /* - * A rejected SET does not mean the server declined it. `commandTimeout` - * gives up client-side while the command may 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. - * - * Reclaiming it is safe because this is the same compare-and-delete - * `releaseLock` uses on the success path: it deletes only while `value` - * still owns the key, so a lock a different holder won in the meantime is - * left alone. Best effort — if Redis is still unreachable the TTL remains - * the backstop, which is exactly the behavior without this cleanup. - */ - await releaseLock(lockKey, value).catch(() => {}) + // 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 } }