Skip to content

Commit 666b4d3

Browse files
authored
fix(knowledge): unschedule a connector whose credential the source rejected and prompt to reconnect (#8158)
* fix(knowledge): unschedule a connector whose credential the source rejected and prompt to reconnect * fix(knowledge): resume by reconnected account across a Slack installation, recheck the rejection before unscheduling, and fail a run that cannot record it
1 parent f492805 commit 666b4d3

14 files changed

Lines changed: 436 additions & 6 deletions

File tree

apps/sim/app/workspace/[workspaceId]/knowledge/[id]/components/connectors-section/connector-recovery.tsx

Lines changed: 9 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,10 @@ import { useEffect, useState } from 'react'
44
import { Chip } from '@sim/emcn'
55
import type { ConnectorData } from '@/lib/api/contracts/knowledge/connectors'
66
import { type ResourceScope, resourceScopeFields } from '@/lib/core/resource-scope'
7-
import { CREDENTIAL_REMOVED_SYNC_ERROR } from '@/lib/knowledge/connectors/sync-limits'
7+
import {
8+
CREDENTIAL_REMOVED_SYNC_ERROR,
9+
CREDENTIAL_REVOKED_SYNC_ERROR,
10+
} from '@/lib/knowledge/connectors/sync-limits'
811
import { getCanonicalScopesForProvider, getProviderIdFromServiceId } from '@/lib/oauth'
912
import { getMissingRequiredScopes } from '@/lib/oauth/utils'
1013
import { ConnectOAuthModal } from '@/app/workspace/[workspaceId]/components/connect-oauth-modal'
@@ -82,7 +85,10 @@ export function ConnectorRecovery({
8285
const docsUrl = isSearchIndex ? connectorDef?.searchDocsUrl : undefined
8386
const credentialRemoved =
8487
connector.lastSyncError === CREDENTIAL_REMOVED_SYNC_ERROR && !connector.credentialId
85-
const pausedTitle = credentialRemoved
88+
const credentialRevoked =
89+
connector.lastSyncError === CREDENTIAL_REVOKED_SYNC_ERROR && Boolean(connector.credentialId)
90+
const reconnectRequired = credentialRemoved || credentialRevoked
91+
const pausedTitle = reconnectRequired
8692
? 'Reconnect to resume syncing'
8793
: 'Sync paused after repeated failures'
8894

@@ -97,7 +103,7 @@ export function ConnectorRecovery({
97103
}
98104
/>
99105
)}
100-
{connector.status === 'disabled' || credentialRemoved ? (
106+
{connector.status === 'disabled' || reconnectRequired ? (
101107
<SettingsResourceRow
102108
title={
103109
!canEdit

apps/sim/connectors/types.ts

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -266,6 +266,7 @@ export const SYNC_SKIP_REASONS = [
266266
'sync_superseded',
267267
'connector_deleted_during_sync',
268268
'credential_missing',
269+
'credential_revoked',
269270
] as const
270271

271272
export type SyncSkipReason = (typeof SYNC_SKIP_REASONS)[number]

apps/sim/lib/credentials/draft-hooks.test.ts

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -103,6 +103,7 @@ describe('handleReconnectCredential', () => {
103103
{ id: 'credential-1', accountId: null, displayName: 'Renamed Gmail' },
104104
])
105105
queueTableRows(schemaMock.credential, [])
106+
queueTableRows(schemaMock.account, [{ providerId: 'gmail', accountId: 'subject-new' }])
106107

107108
await handleReconnectCredential({
108109
draft: { credentialId: 'credential-1' },
@@ -115,6 +116,15 @@ describe('handleReconnectCredential', () => {
115116
expect(mocks.clearDeadFlag).toHaveBeenCalledWith(
116117
getOAuthRefreshCoordinationIdentity('account-new')
117118
)
119+
/** Connectors the rejected credential had unscheduled are due again. */
120+
expect(dbChainMockFns.set).toHaveBeenCalledWith(
121+
expect.objectContaining({
122+
status: 'active',
123+
lastSyncError: null,
124+
consecutiveFailures: 0,
125+
nextSyncAt: new Date('2026-08-14T18:00:00.000Z'),
126+
})
127+
)
118128
expect(auditMockFns.mockRecordAudit).toHaveBeenCalledWith(
119129
expect.objectContaining({
120130
resourceId: 'credential-1',

apps/sim/lib/credentials/draft-hooks.ts

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ import { getPostgresConstraintName, getPostgresErrorCode } from '@sim/utils/erro
66
import { generateId } from '@sim/utils/id'
77
import { and, eq, sql } from 'drizzle-orm'
88
import { deleteOrphanedOAuthAccount } from '@/lib/credentials/deletion'
9+
import { resumeConnectorsAfterCredentialReconnect } from '@/lib/knowledge/connectors/credential-recovery'
910
import { clearOAuthRefreshDeadFlag } from '@/lib/oauth/refresh-coordination'
1011
import { captureServerEvent } from '@/lib/posthog/server'
1112

@@ -75,6 +76,7 @@ export async function handleCreateCredentialFromDraft(params: {
7576
.where(eq(schema.credential.id, existingCredential.id))
7677

7778
await clearOAuthRefreshDeadFlag(accountId)
79+
await resumeConnectorsAfterCredentialReconnect(accountId, now)
7880

7981
recordAudit({
8082
workspaceId: draft.workspaceId,
@@ -209,6 +211,7 @@ export async function handleReconnectCredential(params: {
209211
)
210212

211213
await clearOAuthRefreshDeadFlag(newAccountId)
214+
await resumeConnectorsAfterCredentialReconnect(newAccountId, now)
212215

213216
recordAudit({
214217
workspaceId,

apps/sim/lib/credentials/organization-draft.ts

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ import { OrchestrationError } from '@/lib/core/orchestration/types'
88
import { resourceScopeCondition } from '@/lib/core/resource-scope.server'
99
import { deleteOrphanedOAuthAccount } from '@/lib/credentials/deletion'
1010
import { getCredentialCreationOrganizationContext } from '@/lib/credentials/organization'
11+
import { resumeConnectorsAfterCredentialReconnect } from '@/lib/knowledge/connectors/credential-recovery'
1112
import { clearOAuthRefreshDeadFlag } from '@/lib/oauth/refresh-coordination'
1213

1314
/** Completes the exact draft bound to the authenticated provider callback, rechecking current ownership under membership locks. */
@@ -124,6 +125,7 @@ export async function completeOrganizationCredentialDraft(input: {
124125
}
125126
})
126127
await clearOAuthRefreshDeadFlag(input.accountId)
128+
if (result.reconnected) await resumeConnectorsAfterCredentialReconnect(input.accountId, now)
127129
recordAudit({
128130
actorId: input.userId,
129131
action: result.reconnected
Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,52 @@
1+
/**
2+
* @vitest-environment node
3+
*/
4+
import { dbChainMockFns, queueTableRows, resetDbChainMock, schemaMock } from '@sim/testing'
5+
import { beforeEach, describe, expect, it, vi } from 'vitest'
6+
import { resumeConnectorsAfterCredentialReconnect } from '@/lib/knowledge/connectors/credential-recovery'
7+
import { CREDENTIAL_REVOKED_SYNC_ERROR } from '@/lib/knowledge/connectors/sync-limits'
8+
9+
const RESUMED = {
10+
status: 'active',
11+
lastSyncError: null,
12+
consecutiveFailures: 0,
13+
}
14+
15+
describe('resumeConnectorsAfterCredentialReconnect', () => {
16+
const now = new Date('2026-09-22T20:00:00.000Z')
17+
18+
beforeEach(() => {
19+
vi.clearAllMocks()
20+
resetDbChainMock()
21+
})
22+
23+
it('resumes the connectors of every credential on the reconnected account', async () => {
24+
queueTableRows(schemaMock.account, [
25+
{ providerId: 'confluence', providerAccountId: 'subject-1' },
26+
])
27+
await resumeConnectorsAfterCredentialReconnect('account-1', now)
28+
expect(dbChainMockFns.set).toHaveBeenCalledWith({ ...RESUMED, nextSyncAt: now, updatedAt: now })
29+
const guard = JSON.stringify(dbChainMockFns.where.mock.calls.at(-1))
30+
expect(guard).toContain('knowledgeConnector.credentialId')
31+
expect(guard).toContain(CREDENTIAL_REVOKED_SYNC_ERROR)
32+
const credentials = JSON.stringify(dbChainMockFns.where.mock.calls)
33+
expect(credentials).toContain('"left":"credential.accountId","right":"account-1"')
34+
})
35+
36+
it('resumes across the Slack installation, whose sibling accounts share the repaired chain', async () => {
37+
queueTableRows(schemaMock.account, [
38+
{ providerId: 'slack', providerAccountId: 'TEXAMPLE-usr_U1' },
39+
])
40+
await resumeConnectorsAfterCredentialReconnect('account-1', now)
41+
expect(dbChainMockFns.set).toHaveBeenCalledWith({ ...RESUMED, nextSyncAt: now, updatedAt: now })
42+
const conditions = JSON.stringify(dbChainMockFns.where.mock.calls)
43+
expect(conditions).toContain('"pattern":"TEXAMPLE-%"')
44+
expect(conditions).not.toContain('"left":"credential.accountId","right":"account-1"')
45+
})
46+
47+
it('does nothing for an account that no longer exists', async () => {
48+
queueTableRows(schemaMock.account, [])
49+
await resumeConnectorsAfterCredentialReconnect('account-gone', now)
50+
expect(dbChainMockFns.update).not.toHaveBeenCalled()
51+
})
52+
})
Lines changed: 57 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,57 @@
1+
import { db } from '@sim/db'
2+
import { account, credential, knowledgeConnector } from '@sim/db/schema'
3+
import { and, eq, inArray } from 'drizzle-orm'
4+
import { CREDENTIAL_REVOKED_SYNC_ERROR } from '@/lib/knowledge/connectors/sync-limits'
5+
import { extractSlackTeamId, installationFilter, isSlackProvider } from '@/lib/oauth/slack'
6+
7+
/**
8+
* Puts the connectors a reconnected account had unscheduled back on their schedule.
9+
*
10+
* A sync that finds its credential rejected by the source leaves the connector unscheduled
11+
* with {@link CREDENTIAL_REVOKED_SYNC_ERROR}, since retrying cannot help until someone
12+
* authorizes again. Reauthorizing the account is that moment: every connector on a credential
13+
* of that account still carrying the error is due now, with its failure count cleared. A Slack
14+
* reauthorization repairs the installation's shared token chain, so the connectors on every
15+
* credential of the installation's sibling accounts are due as well. Connectors paused or
16+
* disabled for another reason keep their state, and a connector that already moved on is left
17+
* alone.
18+
*/
19+
export async function resumeConnectorsAfterCredentialReconnect(
20+
accountId: string,
21+
now: Date
22+
): Promise<void> {
23+
const [reconnected] = await db
24+
.select({ providerId: account.providerId, providerAccountId: account.accountId })
25+
.from(account)
26+
.where(eq(account.id, accountId))
27+
.limit(1)
28+
if (!reconnected) return
29+
const slackTeamId = isSlackProvider(reconnected.providerId)
30+
? extractSlackTeamId(reconnected.providerAccountId)
31+
: null
32+
const repairedAccounts = slackTeamId
33+
? inArray(
34+
credential.accountId,
35+
db.select({ id: account.id }).from(account).where(installationFilter(slackTeamId))
36+
)
37+
: eq(credential.accountId, accountId)
38+
await db
39+
.update(knowledgeConnector)
40+
.set({
41+
status: 'active',
42+
lastSyncError: null,
43+
consecutiveFailures: 0,
44+
nextSyncAt: now,
45+
updatedAt: now,
46+
})
47+
.where(
48+
and(
49+
inArray(
50+
knowledgeConnector.credentialId,
51+
db.select({ id: credential.id }).from(credential).where(repairedAccounts)
52+
),
53+
eq(knowledgeConnector.status, 'error'),
54+
eq(knowledgeConnector.lastSyncError, CREDENTIAL_REVOKED_SYNC_ERROR)
55+
)
56+
)
57+
}

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

Lines changed: 127 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
*/
44
import {
55
authOAuthUtilsMock,
6+
authOAuthUtilsMockFns,
67
dbChainMockFns,
78
drizzleOrmMock,
89
flattenMockConditions,
@@ -17,6 +18,7 @@ import { DrizzleQueryError } from 'drizzle-orm/errors'
1718
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
1819
import * as connectorTokens from '@/lib/knowledge/connectors/access-token'
1920
import { executeSync, isConnectorRunnableStatus } from '@/lib/knowledge/connectors/sync-engine'
21+
import { CREDENTIAL_REVOKED_SYNC_ERROR } from '@/lib/knowledge/connectors/sync-limits'
2022
import {
2123
classifySuspectListing,
2224
evaluateListingSafety,
@@ -3007,6 +3009,131 @@ describe('executeSync heartbeats during the listing phase', () => {
30073009
}
30083010
)
30093011

3012+
/** A locked OAuth connector whose token resolution the test controls. */
3013+
function primeOAuthRunUpToToken() {
3014+
const oauthConnector = {
3015+
...CONNECTOR,
3016+
connectorType: 'oauth',
3017+
credentialId: 'cred-1',
3018+
accessMode: 'workspace',
3019+
}
3020+
queueTableRows(schemaMock.knowledgeConnector, [oauthConnector])
3021+
for (let i = 0; i < 20; i++)
3022+
queueTableRows(schemaMock.knowledgeConnector, [
3023+
{ id: 'c-1', connectorArchivedAt: null, connectorDeletedAt: null, kbDeletedAt: null },
3024+
])
3025+
queueTableRows(schemaMock.knowledgeBase, [{ userId: 'u-1', workspaceId: 'ws-1' }])
3026+
dbChainMockFns.returning.mockReset()
3027+
dbChainMockFns.returning.mockResolvedValueOnce([oauthConnector])
3028+
/** The terminal write lands on the row this run still holds. */
3029+
dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'c-1' }])
3030+
const tokenUser = vi
3031+
.spyOn(connectorTokens, 'resolveConnectorTokenUserId')
3032+
.mockResolvedValueOnce('u-1')
3033+
const resolveToken = vi
3034+
.spyOn(connectorTokens, 'resolveConnectorAccessToken')
3035+
.mockResolvedValueOnce(null)
3036+
return () => {
3037+
tokenUser.mockRestore()
3038+
resolveToken.mockRestore()
3039+
}
3040+
}
3041+
3042+
it('unschedules a connector whose credential the source rejected instead of retrying it', async () => {
3043+
const restore = primeOAuthRunUpToToken()
3044+
/** Rejected at token resolution and still rejected when the run records its outcome. */
3045+
authOAuthUtilsMockFns.mockGetCredentialTerminalRefreshError
3046+
.mockResolvedValueOnce('invalid_grant')
3047+
.mockResolvedValueOnce('invalid_grant')
3048+
try {
3049+
const result = await executeSync('c-1', {
3050+
billingAttribution: { workspaceId: 'ws-1' } as never,
3051+
})
3052+
expect(result.skipReason).toBe('credential_revoked')
3053+
expect(result.error).toBeUndefined()
3054+
expect(authOAuthUtilsMockFns.mockGetCredentialTerminalRefreshError).toHaveBeenCalledWith(
3055+
'cred-1'
3056+
)
3057+
expect(dbChainMockFns.set).toHaveBeenCalledWith(
3058+
expect.objectContaining({
3059+
status: 'error',
3060+
nextSyncAt: null,
3061+
lastSyncError: CREDENTIAL_REVOKED_SYNC_ERROR,
3062+
syncLockToken: null,
3063+
syncLockLeaseAt: null,
3064+
})
3065+
)
3066+
expect(dbChainMockFns.set).not.toHaveBeenCalledWith(
3067+
expect.objectContaining({ consecutiveFailures: expect.any(Number) })
3068+
)
3069+
} finally {
3070+
restore()
3071+
}
3072+
})
3073+
3074+
it('takes the failure ladder when the credential was reauthorized while the run was failing', async () => {
3075+
const restore = primeOAuthRunUpToToken()
3076+
/** Rejected at token resolution, repaired by the time the run records its outcome. */
3077+
authOAuthUtilsMockFns.mockGetCredentialTerminalRefreshError
3078+
.mockResolvedValueOnce('invalid_grant')
3079+
.mockResolvedValueOnce(null)
3080+
try {
3081+
const result = await executeSync('c-1', {
3082+
billingAttribution: { workspaceId: 'ws-1' } as never,
3083+
})
3084+
expect(result.skipReason).toBeUndefined()
3085+
expect(result.error).toContain('rejected by the source')
3086+
expect(dbChainMockFns.set).toHaveBeenCalledWith(
3087+
expect.objectContaining({ status: 'error', consecutiveFailures: 1 })
3088+
)
3089+
expect(dbChainMockFns.set).not.toHaveBeenCalledWith(
3090+
expect.objectContaining({ lastSyncError: CREDENTIAL_REVOKED_SYNC_ERROR })
3091+
)
3092+
} finally {
3093+
restore()
3094+
}
3095+
})
3096+
3097+
it('reports a run that could not record the unschedule as failed, not skipped', async () => {
3098+
const restore = primeOAuthRunUpToToken()
3099+
/** Rejected at token resolution and still rejected when the run records its outcome. */
3100+
authOAuthUtilsMockFns.mockGetCredentialTerminalRefreshError
3101+
.mockResolvedValueOnce('invalid_grant')
3102+
.mockResolvedValueOnce('invalid_grant')
3103+
/** The terminal write fails after the lock CAS consumed the first result. */
3104+
dbChainMockFns.returning.mockReset()
3105+
dbChainMockFns.returning.mockResolvedValueOnce([
3106+
{ ...CONNECTOR, connectorType: 'oauth', credentialId: 'cred-1', accessMode: 'workspace' },
3107+
])
3108+
dbChainMockFns.returning.mockRejectedValueOnce(new Error('connection reset'))
3109+
try {
3110+
const result = await executeSync('c-1', {
3111+
billingAttribution: { workspaceId: 'ws-1' } as never,
3112+
})
3113+
expect(result.skipReason).toBeUndefined()
3114+
expect(result.error).toContain('connection reset')
3115+
} finally {
3116+
restore()
3117+
}
3118+
})
3119+
3120+
it('keeps the failure ladder for a credential that resolved no token without a terminal error', async () => {
3121+
const restore = primeOAuthRunUpToToken()
3122+
authOAuthUtilsMockFns.mockGetCredentialTerminalRefreshError.mockResolvedValueOnce(null)
3123+
try {
3124+
const result = await executeSync('c-1', {
3125+
billingAttribution: { workspaceId: 'ws-1' } as never,
3126+
})
3127+
expect(result.skipReason).toBeUndefined()
3128+
expect(result.error).toContain('Failed to obtain access token')
3129+
expect(dbChainMockFns.set).toHaveBeenCalledWith(
3130+
expect.objectContaining({ status: 'error', consecutiveFailures: 1 })
3131+
)
3132+
} finally {
3133+
restore()
3134+
}
3135+
})
3136+
30103137
it.each([
30113138
{ acl: undefined, incomplete: true },
30123139
{ acl: ['invalid-token'], incomplete: true },

0 commit comments

Comments
 (0)