fix(oauth): version-guard slack chain writes against concurrent connects

TheodoreSpeaks · TheodoreSpeaks · commit 3708198f41fa · 2026-07-16T18:25:03.000-07:00
diff --git a/apps/sim/app/api/auth/oauth/utils.test.ts b/apps/sim/app/api/auth/oauth/utils.test.ts
@@ -328,7 +328,7 @@ describe('OAuth Utils', () => {
       expect(redisConfigMockFns.mockAcquireLock.mock.calls[0][0]).toBe(
         'oauth:refresh:slack:T08CM6ZNYBE'
       )
-      expect(redisConfigMockFns.mockAcquireLock.mock.calls[0][2]).toBe(20)
+      expect(redisConfigMockFns.mockAcquireLock.mock.calls[0][2]).toBe(30)
       expect(mockRefreshOAuthToken).toHaveBeenCalledWith('slack', 'live-rt')
       expect(mockSet).toHaveBeenCalledWith(
         expect.objectContaining({ accessToken: 'new-at', refreshToken: 'new-rt' })
@@ -384,6 +384,7 @@ describe('OAuth Utils', () => {
         ok: false,
         errorCode: 'token_revoked',
       })
+      mockSelectChain([])
 
       await expect(refreshTokenIfNeeded('request-id', slackCredential(), 'row-1')).rejects.toThrow(
         'Failed to refresh token'
@@ -396,6 +397,29 @@ describe('OAuth Utils', () => {
         3600
       )
     })
+
+    it('skips the dead flag when the chain moved during the failed refresh', async () => {
+      const fakeRedis = {
+        set: vi.fn().mockResolvedValue('OK'),
+        get: vi.fn().mockResolvedValue(null),
+        del: vi.fn().mockResolvedValue(1),
+      }
+      redisConfigMockFns.mockGetRedisClient.mockReturnValue(fakeRedis)
+      mockSelectOrderedChain([
+        { accessToken: 'stale-at', refreshToken: 'live-rt', accessTokenExpiresAt: past },
+      ])
+      mockRefreshOAuthToken.mockResolvedValueOnce({
+        ok: false,
+        errorCode: 'token_revoked',
+      })
+      mockSelectChain([{ moved: new Date() }])
+
+      await expect(refreshTokenIfNeeded('request-id', slackCredential(), 'row-1')).rejects.toThrow(
+        'Failed to refresh token'
+      )
+
+      expect(fakeRedis.set).not.toHaveBeenCalled()
+    })
   })
 
   describe('resolveServiceAccountToken', () => {
diff --git a/apps/sim/app/api/auth/oauth/utils.ts b/apps/sim/app/api/auth/oauth/utils.ts
@@ -31,6 +31,7 @@ import {
   extractSlackTeamId,
   fanOutSlackTokenChain,
   getFreshestSlackChain,
+  hasSlackChainMoved,
   isSlackProvider,
 } from '@/lib/oauth/slack'
 import {
@@ -673,12 +674,15 @@ interface CoalescedRefreshOptions {
 /**
  * Slack lock budgets sized past `TOKEN_REFRESH_TIMEOUT_MS` (15s) in
  * lib/oauth/oauth.ts: installation-keyed locks make every sibling row's request
- * a follower of one refresh. The lock TTL must not expire under a live refresh
- * (15s provider call plus DB reads and the fan-out write), and followers poll
- * for the lock's full lifetime so a slow-but-successful refresh is still
- * observed rather than reported as a failure.
+ * a follower of one refresh, so the TTL covers the provider call plus generous
+ * headroom for the surrounding DB reads and the fan-out write, and followers
+ * poll for the lock's full lifetime so a slow-but-successful refresh is still
+ * observed rather than reported as a failure. These budgets are latency knobs,
+ * not correctness guarantees — chain integrity under lock expiry or unlocked
+ * concurrent writers is enforced by the version-guarded fan-out
+ * (`ifChainUnchangedSince` in lib/oauth/slack.ts).
  */
-const SLACK_LOCK_TTL_SEC = 20
+const SLACK_LOCK_TTL_SEC = 30
 const SLACK_FOLLOWER_MAX_WAIT_MS = SLACK_LOCK_TTL_SEC * 1000
 
 async function performCoalescedRefresh({
@@ -727,23 +731,29 @@ async function performCoalescedRefresh({
       onLeader: async () => {
         try {
           let refreshTokenToUse = refreshToken
+          let slackChainVersion: Date | null = null
           if (slackTeamId) {
             const freshest = await getFreshestSlackChain(slackTeamId)
             if (!freshest) {
               throw new Error(
                 `No refresh-capable account row found for Slack installation ${slackTeamId}`
               )
             }
+            slackChainVersion = freshest.chainVersion
             if (
               freshest.accessToken &&
               freshest.accessTokenExpiresAt &&
               freshest.accessTokenExpiresAt > new Date()
             ) {
-              await fanOutSlackTokenChain(slackTeamId, {
-                accessToken: freshest.accessToken,
-                refreshToken: freshest.refreshToken,
-                accessTokenExpiresAt: freshest.accessTokenExpiresAt,
-              })
+              await fanOutSlackTokenChain(
+                slackTeamId,
+                {
+                  accessToken: freshest.accessToken,
+                  refreshToken: freshest.refreshToken,
+                  accessTokenExpiresAt: freshest.accessTokenExpiresAt,
+                },
+                { ifChainUnchangedSince: freshest.chainVersion }
+              )
               logger.info('Reused freshest Slack installation token', logContext)
               return freshest.accessToken
             }
@@ -758,19 +768,34 @@ async function performCoalescedRefresh({
               errorCode: result.errorCode,
             })
             if (result.errorCode && isTerminalRefreshError(result.errorCode)) {
-              await markCredentialDead(scopeKey, result.errorCode)
+              // A refresh that lost a race with a concurrent connect fails with
+              // a revoked/rotated-out token even though the installation just
+              // got a live chain — dead-flagging then would take down a healthy
+              // credential for an hour.
+              if (
+                slackChainVersion &&
+                (await hasSlackChainMoved(slackTeamId!, slackChainVersion))
+              ) {
+                logger.info('Skipping dead flag: Slack chain moved during refresh', logContext)
+              } else {
+                await markCredentialDead(scopeKey, result.errorCode)
+              }
             }
             return null
           }
 
           const accessTokenExpiresAt = new Date(Date.now() + result.expiresIn * 1000)
 
           if (slackTeamId) {
-            await fanOutSlackTokenChain(slackTeamId, {
-              accessToken: result.accessToken,
-              refreshToken: result.refreshToken || refreshTokenToUse,
-              accessTokenExpiresAt,
-            })
+            await fanOutSlackTokenChain(
+              slackTeamId,
+              {
+                accessToken: result.accessToken,
+                refreshToken: result.refreshToken || refreshTokenToUse,
+                accessTokenExpiresAt,
+              },
+              { ifChainUnchangedSince: slackChainVersion ?? undefined }
+            )
           } else {
             const updateData: Record<string, unknown> = {
               accessToken: result.accessToken,
diff --git a/apps/sim/lib/oauth/slack.ts b/apps/sim/lib/oauth/slack.ts
@@ -1,6 +1,6 @@
 import { db } from '@sim/db'
 import { account } from '@sim/db/schema'
-import { and, eq, isNotNull, like, sql } from 'drizzle-orm'
+import { and, eq, gt, isNotNull, like, max, sql } from 'drizzle-orm'
 
 /**
  * Slack bot tokens belong to the installation (team × app), not to the OAuth
@@ -40,13 +40,31 @@ interface SlackTokenChain {
   accessTokenExpiresAt: Date | null
 }
 
+interface FanOutOptions {
+  /**
+   * Version guard: skip the write entirely when any sibling row was updated
+   * after this timestamp. A refresh leader holds its chain snapshot across a
+   * multi-second provider call while a concurrent OAuth connect (whose
+   * `account.create.after` fan-out takes no lock) may land the newly issued
+   * chain first — an unconditional write would then overwrite the live chain
+   * with a stale one. Omit for connect-time fan-out, whose chain is by
+   * definition the newest.
+   */
+  ifChainUnchangedSince?: Date
+}
+
 /**
  * Writes a token chain to every account row of a Slack installation. Rotation
  * revokes whatever the sibling rows were holding, so a successful refresh or a
  * fresh connect must overwrite all copies or the stale ones fail with
  * `token_revoked` at call time.
  */
-export async function fanOutSlackTokenChain(teamId: string, chain: SlackTokenChain): Promise<void> {
+export async function fanOutSlackTokenChain(
+  teamId: string,
+  chain: SlackTokenChain,
+  options?: FanOutOptions
+): Promise<void> {
+  const since = options?.ifChainUnchangedSince
   await db
     .update(account)
     .set({
@@ -55,13 +73,40 @@ export async function fanOutSlackTokenChain(teamId: string, chain: SlackTokenCha
       ...(chain.refreshToken ? { refreshToken: chain.refreshToken } : {}),
       updatedAt: new Date(),
     })
-    .where(installationFilter(teamId))
+    .where(
+      and(
+        installationFilter(teamId),
+        since
+          ? sql`NOT EXISTS (SELECT 1 FROM ${account} sibling WHERE sibling.provider_id = 'slack' AND sibling.account_id LIKE ${`${teamId}-%`} AND sibling.updated_at > ${since})`
+          : undefined
+      )
+    )
+}
+
+/**
+ * True when any account row of the installation was updated after `since` —
+ * i.e. another writer (a fresh connect or a competing refresh) landed a newer
+ * chain while the caller was working from an older snapshot.
+ */
+export async function hasSlackChainMoved(teamId: string, since: Date): Promise<boolean> {
+  const [row] = await db
+    .select({ moved: max(account.updatedAt) })
+    .from(account)
+    .where(and(installationFilter(teamId), gt(account.updatedAt, since)))
+    .limit(1)
+  return row?.moved != null
 }
 
 interface FreshestSlackChain {
   accessToken: string | null
   refreshToken: string
   accessTokenExpiresAt: Date | null
+  /**
+   * Max `updated_at` across the installation's rows at read time — the version
+   * guard passed back into {@link fanOutSlackTokenChain} / consulted via
+   * {@link hasSlackChainMoved} after the provider round-trip.
+   */
+  chainVersion: Date
 }
 
 /**
@@ -76,6 +121,9 @@ export async function getFreshestSlackChain(teamId: string): Promise<FreshestSla
       accessToken: account.accessToken,
       refreshToken: account.refreshToken,
       accessTokenExpiresAt: account.accessTokenExpiresAt,
+      chainVersion: sql<
+        Date | string | null
+      >`(SELECT max(sibling.updated_at) FROM ${account} sibling WHERE sibling.provider_id = 'slack' AND sibling.account_id LIKE ${`${teamId}-%`})`,
     })
     .from(account)
     .where(and(installationFilter(teamId), isNotNull(account.refreshToken)))
@@ -87,5 +135,6 @@ export async function getFreshestSlackChain(teamId: string): Promise<FreshestSla
     accessToken: row.accessToken,
     refreshToken: row.refreshToken,
     accessTokenExpiresAt: row.accessTokenExpiresAt,
+    chainVersion: row.chainVersion ? new Date(row.chainVersion) : new Date(0),
   }
 }
