fix(oauth): guard revocation records against in-flight reconnects

- Record a revocation only when the row's updated_at has not moved since
  the leader read it, so a reconnect that keeps the same refresh token
  while a rejected refresh is in flight is not undone.
- Reread the row under the refresh lock, so a revocation another process
  just recorded is honored instead of calling the provider again.
- A re-probe whose transient failure meets a moved chain reports no
  revocation.
- Slack fan-out clears the record only for a chain the provider just
  issued, not when re-spreading a stored one.
- A reconnect whose cleanup fails now fails the callback.
- Revocation rejections log at WARN in oauth.ts too; the pure refresh
  error classifiers move to refresh-error-codes.ts so that client-reachable
  module can use them.
- The connector path leaves the revocation log to executeSync.

Author: waleedlatif1

apps/sim/app/api/auth/oauth/utils.test.ts

@@ -100,6 +100,8 @@ describe('OAuth Utils', () => {
     }
 
     it('locks per installation and refreshes with the freshest sibling refresh token', async () => {
+      // The leader rereads the row under the lock.
+      mockSelectChain([])
       mockSelectOrderedChain([
         { accessToken: 'stale-at', refreshToken: 'live-rt', accessTokenExpiresAt: past },
       ])
@@ -127,6 +129,8 @@ describe('OAuth Utils', () => {
     })
 
     it('returns the freshest sibling token without refreshing when it is still valid', async () => {
+      // The leader rereads the row under the lock.
+      mockSelectChain([])
       mockSelectOrderedChain([
         { accessToken: 'sibling-at', refreshToken: 'live-rt', accessTokenExpiresAt: future },
       ])
@@ -148,6 +152,8 @@ describe('OAuth Utils', () => {
         del: vi.fn().mockResolvedValue(1),
       }
       redisConfigMockFns.mockGetRedisClient.mockReturnValue(fakeRedis)
+      // The leader rereads the row under the lock.
+      mockSelectChain([])
       mockSelectOrderedChain([
         { accessToken: 'stale-at', refreshToken: 'live-rt', accessTokenExpiresAt: past },
       ])
@@ -177,6 +183,8 @@ describe('OAuth Utils', () => {
         del: vi.fn().mockResolvedValue(1),
       }
       redisConfigMockFns.mockGetRedisClient.mockReturnValue(fakeRedis)
+      // The leader rereads the row under the lock.
+      mockSelectChain([])
       mockSelectOrderedChain([
         { accessToken: 'stale-at', refreshToken: 'live-rt', accessTokenExpiresAt: past },
       ])

apps/sim/lib/auth/auth.ts

@@ -560,11 +560,15 @@ export const auth = betterAuth({
                 // proves the installation has live tokens, and a fan-out
                 // failure must not leave the hour-long flag blocking refreshes.
                 await clearOAuthRefreshDeadFlag(`slack:${teamId}`)
-                await fanOutSlackTokenChain(teamId, {
-                  accessToken: account.accessToken,
-                  refreshToken: account.refreshToken ?? null,
-                  accessTokenExpiresAt: account.accessTokenExpiresAt ?? null,
-                })
+                await fanOutSlackTokenChain(
+                  teamId,
+                  {
+                    accessToken: account.accessToken,
+                    refreshToken: account.refreshToken ?? null,
+                    accessTokenExpiresAt: account.accessTokenExpiresAt ?? null,
+                  },
+                  { freshlyIssued: true }
+                )
                 logger.info('[account.create.after] Propagated Slack installation token chain', {
                   userId: account.userId,
                   teamId,
@@ -711,15 +715,9 @@ export const auth = betterAuth({
         after: async (account, context) => {
           const path = context?.path
           if (!path?.startsWith('/oauth2/callback/') && !path?.startsWith('/callback/')) return
-          try {
-            await clearOAuthRefreshFailure(account.id)
-          } catch (error) {
-            logger.error('[account.update.after] Failed to clear recorded refresh failures', {
-              accountId: account.id,
-              providerId: account.providerId,
-              error,
-            })
-          }
+          // Fails the callback like the draft hooks do, so a reconnect never reports success
+          // while the old revocation still blocks the credential.
+          await clearOAuthRefreshFailure(account.id)
         },
       },
     },

apps/sim/lib/auth/connectors/managed-oauth.ts

@@ -9,8 +9,8 @@ import { readResponseJsonWithLimit } from '@/lib/core/utils/stream-limits'
 import { getDocusignOAuthUrl } from '@/lib/oauth/docusign'
 import { verifyGitHubRepositoriesIdentity } from '@/lib/oauth/github-repositories'
 import { deriveMicrosoftEmailVerified, mapMicrosoftProfileToUser } from '@/lib/oauth/microsoft'
+import { isTerminalRefreshError } from '@/lib/oauth/refresh-error-codes'
 import { SALESFORCE_LOGIN_HOSTS } from '@/lib/oauth/salesforce'
-import { isTerminalRefreshError } from '@/lib/oauth/terminal-errors'
 import { getCanonicalScopesForProvider, isScopeSatisfiedBy } from '@/lib/oauth/utils'
 import { MONDAY_API_URL, MONDAY_API_VERSION } from '@/tools/monday/utils'
 

apps/sim/lib/knowledge/connectors/sync-engine.ts

@@ -96,7 +96,7 @@ import {
 import { hardDeleteDocuments } from '@/lib/knowledge/documents/service'
 import { getRetryAfterMs, isRateLimitError } from '@/lib/knowledge/documents/utils'
 import { getCredentialTerminalRefreshError } from '@/lib/oauth/credential-service'
-import { isCredentialRevocationError } from '@/lib/oauth/terminal-errors'
+import { isCredentialRevocationError } from '@/lib/oauth/refresh-error-codes'
 import { connectorHasAuthSource } from '@/connectors/auth'
 import { CONNECTOR_REGISTRY } from '@/connectors/registry.server'
 import type {
@@ -823,19 +823,15 @@ async function resolveAccessToken(
       connectorConfig.auth.mode === 'oauth' && connector.credentialId
         ? await getCredentialRevocationError(connector.credentialId)
         : null
-    const logContext = {
-      credentialId: connector.credentialId,
-      userId,
-      authMode: connectorConfig.auth.mode,
-    }
+    // executeSync logs the revocation once when it unschedules the connector.
     if (revocationError && connector.credentialId) {
-      logger.warn(`[${requestId}] Connector credential was revoked by the source`, {
-        ...logContext,
-        errorCode: revocationError,
-      })
       throw new ConnectorCredentialRevokedError(connector.credentialId, revocationError)
     }
-    logger.error(`[${requestId}] Connector credential resolved no access token`, logContext)
+    logger.error(`[${requestId}] Connector credential resolved no access token`, {
+      credentialId: connector.credentialId,
+      userId,
+      authMode: connectorConfig.auth.mode,
+    })
     throw new Error(`Failed to obtain access token for credential ${connector.credentialId}`)
   }
 

apps/sim/lib/oauth/credential-revocation.integration.ts

@@ -332,6 +332,31 @@ describe('OAuth refresh-token revocation against PostgreSQL and Redis', () => {
     expect(providerRequests).toBe(2)
   })
 
+  it('keeps a reconnect that lands while a rejected refresh is in flight', async () => {
+    pausedProvider = { reached: createDeferred<void>(), release: createDeferred<void>() }
+    const inFlight = resolveToken().catch((error: unknown) => error)
+    await pausedProvider.reached.promise
+    await db
+      .update(account)
+      .set({
+        accessToken: 'relinked-access',
+        accessTokenExpiresAt: new Date(Date.now() + 3_600_000),
+        updatedAt: new Date(),
+      })
+      .where(eq(account.id, accountId))
+    await handleReconnectCredential({
+      draft: { credentialId },
+      newAccountId: accountId,
+      workspaceId,
+      userId,
+      now: new Date(),
+    })
+    pausedProvider.release.resolve()
+
+    expect(await inFlight).not.toBeInstanceOf(CredentialRevokedError)
+    expect((await storedAccount()).refreshRevokedAt).toBeNull()
+  })
+
   it('answers a waiting refresh from the revocation another process recorded', async () => {
     await expect(resolveToken()).rejects.toBeInstanceOf(CredentialRevokedError)
     const { refreshRevokedAt, refreshRevokedCode, refreshRevokedTokenHash } = await storedAccount()

apps/sim/lib/oauth/credential-service.test.ts

@@ -57,9 +57,12 @@ vi.mock('@/lib/oauth/slack', () => ({
 
 vi.mock('@/lib/oauth/terminal-errors', () => ({
   getRecentTerminalError: hoisted.getRecentTerminalError,
+  markCredentialDead: vi.fn(),
+}))
+
+vi.mock('@/lib/oauth/refresh-error-codes', () => ({
   isCredentialRevocationError: vi.fn(() => false),
   isTerminalRefreshError: vi.fn(() => false),
-  markCredentialDead: vi.fn(),
 }))
 
 import {
@@ -74,8 +77,9 @@ import {
 import { isInstagramProvider, shouldProactivelyRefreshInstagramToken } from '@/lib/oauth/instagram'
 import { isMicrosoftProvider } from '@/lib/oauth/microsoft'
 import { getOAuthRefreshCoordinationIdentity } from '@/lib/oauth/refresh-coordination'
+import { isTerminalRefreshError } from '@/lib/oauth/refresh-error-codes'
 import { fanOutSlackTokenChain } from '@/lib/oauth/slack'
-import { isTerminalRefreshError, markCredentialDead } from '@/lib/oauth/terminal-errors'
+import { markCredentialDead } from '@/lib/oauth/terminal-errors'
 import { GOOGLE_SERVICE_ACCOUNT_PROVIDER_ID } from '@/lib/oauth/types'
 
 const serviceLogger = getMockLogger('OAuthCredentialService')
@@ -505,6 +509,7 @@ describe('OAuth access-token refresh headroom', () => {
   it('returns no token when the rotation write finds the account gone', async () => {
     queueCredentialAccount(createOAuthAccount())
     dbChainMockFns.returning.mockResolvedValueOnce([])
+    queueTableRows(account, [createOAuthAccount()])
     queueTableRows(account, [])
     await expect(
       resolveCredentialTokenBundle(RAW_CREDENTIAL_ID, RAW_USER_ID, 'test')
@@ -515,6 +520,7 @@ describe('OAuth access-token refresh headroom', () => {
     queueCredentialAccount(createOAuthAccount())
     vi.mocked(isTerminalRefreshError).mockReturnValue(true)
     mocks.refreshOAuthToken.mockResolvedValue({ ok: false, errorCode: 'invalid_client' })
+    queueTableRows(account, [createOAuthAccount()])
     queueTableRows(account, [
       {
         ...createOAuthAccount(3_600_000),
@@ -533,6 +539,7 @@ describe('OAuth access-token refresh headroom', () => {
     vi.mocked(isTerminalRefreshError).mockReturnValue(true)
     mocks.refreshOAuthToken.mockResolvedValue({ ok: false, errorCode: 'invalid_client' })
     queueTableRows(account, [createOAuthAccount()])
+    queueTableRows(account, [createOAuthAccount()])
     await expect(
       resolveCredentialTokenBundle(RAW_CREDENTIAL_ID, RAW_USER_ID, 'test')
     ).resolves.toBeNull()
@@ -543,6 +550,7 @@ describe('OAuth access-token refresh headroom', () => {
     queueCredentialAccount(createOAuthAccount())
     /** Another writer rotated first: no row still holds the token this refresh started from. */
     dbChainMockFns.returning.mockResolvedValueOnce([])
+    queueTableRows(account, [createOAuthAccount()])
     queueTableRows(account, [{ ...createOAuthAccount(3_600_000), accessToken: 'winner-token' }])
     await expect(
       resolveCredentialTokenBundle(RAW_CREDENTIAL_ID, RAW_USER_ID, 'test')
@@ -578,7 +586,9 @@ describe('OAuth access-token refresh headroom', () => {
         expect.objectContaining({
           accessToken: refresh ? 'refreshed-access-token' : 'installation-access-token',
         }),
-        { ifChainUnchangedSince: chainVersion }
+        refresh
+          ? { ifChainUnchangedSince: chainVersion, freshlyIssued: true }
+          : { ifChainUnchangedSince: chainVersion }
       )
     }
   )