fix(redis): reclaim a distributed lock that a timed-out acquire may have taken (#6864)

* fix(redis): reclaim a lock a timed-out acquire may have taken

`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) <noreply@anthropic.com>

* fix(redis): make the timed-out-acquire reclaim opt-in per caller

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) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

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()

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' },

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 },

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()

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(
           {

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()

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(

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'

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<boolean> {
   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
+  }
 }
 
 /**