Function runWithRedisMutex

apps/sim/lib/mcp/oauth/storage.ts:312–366

(
  lockKey: string,
  rowId: string,
  fn: () => Promise<T>
)

}

async function runWithRedisMutex<T>(
  lockKey: string,
  rowId: string,
  fn: () => Promise<T>
): Promise<T> {
  const ownerToken = generateShortId()
  const deadline = Date.now() + REFRESH_MAX_WAIT_MS

  while (true) {
    let acquired = false
    try {
      acquired = await acquireLock(lockKey, ownerToken, REFRESH_LOCK_TTL_SEC)
    } catch (error) {
      logger.warn('Redis unavailable, running OAuth flow uncoordinated', {
        rowId,
        error: toError(error).message,
      })
      return fn()
    }

    if (acquired) {
      const watchdog = setInterval(() => {
        extendLock(lockKey, ownerToken, REFRESH_LOCK_TTL_SEC).catch((error) => {
          logger.warn('Refresh lock extend failed', {
            rowId,
            error: toError(error).message,
          })
        })
      }, REFRESH_LOCK_EXTEND_INTERVAL_MS)
      try {
        return await fn()
      } finally {
        clearInterval(watchdog)
        await releaseLock(lockKey, ownerToken).catch((error) => {
          logger.warn('Refresh lock release failed (will expire via TTL)', {
            rowId,
            error: toError(error).message,
          })
        })
      }
    }

    if (Date.now() >= deadline) {
      // Lock still held by another process AND its watchdog is keeping it
      // alive — falling open would let us refresh concurrently and race the
      // rotating refresh token. Throw and let the caller decide whether to
      // retry; the Redis-down path remains the only branch that runs `fn()`
      // uncoordinated (no coordination available there).
      throw new Error(
        `MCP OAuth refresh lock for ${rowId} held longer than ${REFRESH_MAX_WAIT_MS}ms`
      )
    }
    await sleep(REFRESH_POLL_INTERVAL_MS)
  }
}