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>

Author: icecrasher321

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

@@ -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 () => {

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<boolean> {
   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
   }
 }