fix(connectors): honor provider retry deadlines (#7211)

* fix(connectors): honor provider retry deadlines

* fix(connectors): validate retry response lifecycles

Author: waleedlatif1

apps/sim/lib/atlassian/discovery.test.ts

@@ -205,6 +205,30 @@ describe('resolveAtlassianCloudId', () => {
   it('rejects when the token can see no sites', async () => {
     fetchMock.mockResolvedValue(sites([]))
 
-    await expect(resolveAtlassianCloudId(options())).rejects.toThrow('No Jira resources found')
+    await expect(resolveAtlassianCloudId(options())).rejects.toThrow(
+      'No Jira sites are accessible to this credential. Reconnect the credential and grant access to the configured Atlassian site.'
+    )
+  })
+
+  it('distinguishes a malformed discovery payload from an empty site grant', async () => {
+    fetchMock.mockResolvedValue(createMockResponse({ json: { id: CLOUD_ID, url: SITE } }))
+
+    await expect(resolveAtlassianCloudId(options())).rejects.toThrow(
+      'Invalid Jira accessible-resources response'
+    )
+  })
+
+  it.each([
+    [{ url: SITE }],
+    [{ id: CLOUD_ID }],
+    [{ id: '', url: SITE }],
+    [{ id: CLOUD_ID, url: '' }],
+    [null],
+  ])('rejects malformed resource entries in an otherwise valid array', async (resources) => {
+    fetchMock.mockResolvedValue(createMockResponse({ json: resources }))
+
+    await expect(resolveAtlassianCloudId(options())).rejects.toThrow(
+      'Invalid Jira accessible-resources response'
+    )
   })
 })

apps/sim/lib/atlassian/discovery.ts

@@ -102,6 +102,17 @@ interface AccessibleResource {
   url: string
 }
 
+function isAccessibleResource(value: unknown): value is AccessibleResource {
+  if (typeof value !== 'object' || value === null) return false
+  const resource = value as Record<string, unknown>
+  return (
+    typeof resource.id === 'string' &&
+    resource.id.trim().length > 0 &&
+    typeof resource.url === 'string' &&
+    resource.url.trim().length > 0
+  )
+}
+
 interface ResolveAtlassianCloudIdOptions {
   domain: string
   accessToken: string
@@ -203,21 +214,26 @@ export function selectAtlassianCloudId(
   domain: string,
   product: string
 ): string {
-  if (!Array.isArray(resources) || resources.length === 0) {
-    throw new Error(`No ${product} resources found`)
+  if (!Array.isArray(resources) || !resources.every(isAccessibleResource)) {
+    throw new Error(`Invalid ${product} accessible-resources response`)
+  }
+
+  if (resources.length === 0) {
+    throw new Error(
+      `No ${product} sites are accessible to this credential. ` +
+        'Reconnect the credential and grant access to the configured Atlassian site.'
+    )
   }
 
   const siteUrl = normalizeAtlassianSiteUrl(domain)
-  const match = (resources as AccessibleResource[]).find(
-    (r) => normalizeAtlassianSiteUrl(r.url) === siteUrl
-  )
+  const match = resources.find((r) => normalizeAtlassianSiteUrl(r.url) === siteUrl)
   if (match) return match.id
 
-  if (resources.length === 1) return (resources as AccessibleResource[])[0].id
+  if (resources.length === 1) return resources[0].id
 
   throw new Error(
     `Could not match ${product} domain "${domain}" to any accessible resource. ` +
-      `Available sites: ${(resources as AccessibleResource[]).map((r) => r.url).join(', ')}`
+      `Available sites: ${resources.map((r) => r.url).join(', ')}`
   )
 }
 

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

@@ -2215,6 +2215,30 @@ describe('buildSyncFailureUpdate', () => {
     expect(buildSyncFailureUpdate(now, undefined, 'boom').nextSyncAt).toEqual(minutesAfter(30))
   })
 
+  it('does not schedule before a longer provider retry deadline', async () => {
+    const { buildSyncFailureUpdate } = await import('@/lib/knowledge/connectors/sync-engine')
+
+    expect(buildSyncFailureUpdate(now, 0, 'rate limited', 45 * 60 * 1000).nextSyncAt).toEqual(
+      minutesAfter(45)
+    )
+  })
+
+  it('does not let a shorter provider delay weaken the failure backoff', async () => {
+    const { buildSyncFailureUpdate } = await import('@/lib/knowledge/connectors/sync-engine')
+
+    expect(buildSyncFailureUpdate(now, 0, 'rate limited', 5 * 60 * 1000).nextSyncAt).toEqual(
+      minutesAfter(30)
+    )
+  })
+
+  it('caps an unreasonable provider delay at the existing one-day retry ceiling', async () => {
+    const { buildSyncFailureUpdate } = await import('@/lib/knowledge/connectors/sync-engine')
+
+    expect(
+      buildSyncFailureUpdate(now, 0, 'rate limited', 30 * 24 * 60 * 60 * 1000).nextSyncAt
+    ).toEqual(minutesAfter(24 * 60))
+  })
+
   it('disables exactly at the threshold, not before it', async () => {
     const { buildSyncFailureUpdate } = await import('@/lib/knowledge/connectors/sync-engine')
     const { MAX_CONSECUTIVE_FAILURES } = await import('@/lib/knowledge/connectors/sync-limits')

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

@@ -35,6 +35,7 @@ import { getInternalApiBaseUrl } from '@/lib/core/utils/urls'
 import { resolveCredentialTokenIdentity } from '@/lib/credentials/access'
 import {
   CONNECTOR_AUTO_DISABLED_ERROR,
+  CONNECTOR_FAILURE_BACKOFF_CAP_MINUTES,
   connectorFailureBackoffMinutes,
   MAX_CONSECUTIVE_FAILURES,
   SYNC_LOCK_HEARTBEAT_INTERVAL_MS,
@@ -52,6 +53,7 @@ import {
   MAX_PROCESSING_ATTEMPTS,
   QUEUED_DISPATCH_GRACE_MS,
 } from '@/lib/knowledge/documents/types'
+import { getRetryAfterMs } from '@/lib/knowledge/documents/utils'
 import { refreshAccessTokenIfNeeded } from '@/lib/oauth/credential-service'
 import { StorageService } from '@/lib/uploads'
 import { buildStorageKeySegment } from '@/lib/uploads/core/storage-key'
@@ -1321,22 +1323,32 @@ export function buildReconciliationHoldNotice(
  * it applies need to be assertable without standing up the whole sync. The
  * in-process ladder here and the reaper's SQL ladder must agree — they are two
  * writers of one policy, both sourced from
- * {@link connectorFailureBackoffMinutes}.
+ * {@link connectorFailureBackoffMinutes}. A validated provider retry delay is
+ * an additional lower bound, capped at the same one-day ceiling: a short hint
+ * cannot weaken the failure ladder, while an untrusted extreme value cannot
+ * pin the connector indefinitely.
  */
 export function buildSyncFailureUpdate(
   now: Date,
   previousFailures: number | null | undefined,
-  errorMessage: string
+  errorMessage: string,
+  retryAfterMs?: number
 ) {
   const failures = (previousFailures ?? 0) + 1
   const disabled = failures >= MAX_CONSECUTIVE_FAILURES
+  const failureBackoffMs = connectorFailureBackoffMinutes(failures) * 60 * 1000
+  const maximumBackoffMs = CONNECTOR_FAILURE_BACKOFF_CAP_MINUTES * 60 * 1000
+  const providerBackoffMs =
+    typeof retryAfterMs === 'number' && Number.isFinite(retryAfterMs) && retryAfterMs > 0
+      ? Math.min(retryAfterMs, maximumBackoffMs)
+      : 0
 
   return {
     status: (disabled ? 'disabled' : 'error') as 'disabled' | 'error',
     lastSyncError: disabled ? CONNECTOR_AUTO_DISABLED_ERROR : errorMessage,
     nextSyncAt: disabled
       ? null
-      : new Date(now.getTime() + connectorFailureBackoffMinutes(failures) * 60 * 1000),
+      : new Date(now.getTime() + Math.max(failureBackoffMs, providerBackoffMs)),
     consecutiveFailures: failures,
     // Releases the lock so a stale token can never match a later run, and closes
     // its lease so the reaper is not left waiting out a TTL on a finished run.
@@ -3160,15 +3172,25 @@ export async function executeSync(
     }
 
     const errorMessage = toError(error).message
-    logger.error('Sync failed', { connectorId, error: errorMessage })
+    const retryAfterMs = getRetryAfterMs(error)
+    logger.error('Sync failed', {
+      connectorId,
+      error: errorMessage,
+      ...(retryAfterMs === undefined ? {} : { retryAfterMs }),
+    })
 
     try {
       await completeSyncLog(syncLogId, 'failed', result, { errorMessage })
 
       const failureUpdate =
         error instanceof ConnectorSyncCapacityError
           ? buildSyncCapacityUpdate(new Date(), connector.consecutiveFailures, errorMessage)
-          : buildSyncFailureUpdate(new Date(), connector.consecutiveFailures, errorMessage)
+          : buildSyncFailureUpdate(
+              new Date(),
+              connector.consecutiveFailures,
+              errorMessage,
+              retryAfterMs
+            )
 
       if (failureUpdate.status === 'disabled') {
         logger.warn('Connector disabled after repeated failures', {

apps/sim/lib/knowledge/documents/secure-fetch.server.ts

@@ -4,12 +4,9 @@ import {
   secureFetchWithValidation,
 } from '@/lib/core/security/input-validation.server'
 import {
-  attachRetryHeaders,
-  type HTTPError,
+  createRetryableHttpError,
   isRetryableError,
   type RetryOptions,
-  readBoundedHttpErrorBody,
-  resolveRetryDelayMs,
   retryWithExponentialBackoff,
 } from '@/lib/knowledge/documents/utils'
 
@@ -56,17 +53,7 @@ export async function secureFetchWithRetry(
      * limit) use instead.
      */
     if (!response.ok && isRetryableError({ status: response.status, headers: response.headers })) {
-      const errorText = await readBoundedHttpErrorBody(response)
-      const error: HTTPError = new Error(`HTTP ${response.status} - ${errorText}`)
-      error.status = response.status
-      attachRetryHeaders(error, response.headers)
-
-      const waitMs = resolveRetryDelayMs(response.headers)
-      if (waitMs !== undefined) {
-        error.retryAfterMs = waitMs
-      }
-
-      throw error
+      throw await createRetryableHttpError(response)
     }
 
     return response

apps/sim/lib/knowledge/documents/utils.test.ts

@@ -14,6 +14,7 @@ vi.mock('@/lib/core/security/input-validation.server', () => ({
 import { secureFetchWithRetry } from './secure-fetch.server'
 import {
   fetchWithRetry,
+  getRetryAfterMs,
   type HTTPError,
   hasRateLimitEvidence,
   isRetryableError,
@@ -535,13 +536,37 @@ describe('fetchWithRetry rate-limit handling', () => {
       .mockResolvedValueOnce(response(200))
     globalThis.fetch = fetchMock
 
-    await expect(fetchWithRetry('https://api.github.com/repos', {}, FAST_RETRY)).rejects.toThrow(
-      'HTTP 403'
+    const error = await fetchWithRetry('https://api.github.com/repos', {}, FAST_RETRY).then(
+      () => undefined,
+      (caught) => caught as Error
     )
 
+    expect(error?.message).toBe('HTTP 403 - upstream rate limit exceeded')
+    expect(getRetryAfterMs(error)).toBeGreaterThan(899_000)
+    expect(getRetryAfterMs(error)).toBeLessThanOrEqual(900_000)
     expect(fetchMock).toHaveBeenCalledTimes(1)
   })
 
+  it('cancels an omitted rate-limit response body before throwing', async () => {
+    let cancelled = false
+    const body = new ReadableStream<Uint8Array>({
+      cancel() {
+        cancelled = true
+      },
+    })
+    globalThis.fetch = vi.fn().mockResolvedValue(
+      new Response(body, {
+        status: 429,
+        headers: { 'retry-after': '900' },
+      })
+    )
+
+    await expect(
+      fetchWithRetry('https://api.github.com/repos', {}, { ...FAST_RETRY, maxRetries: 0 })
+    ).rejects.toThrow('HTTP 429 - upstream rate limit exceeded')
+    expect(cancelled).toBe(true)
+  })
+
   it('waits until an admitted x-rate-limit-reset instant before retrying', async () => {
     vi.useFakeTimers()
     const now = 1_700_000_000_000
@@ -601,6 +626,20 @@ describe('fetchWithRetry rate-limit handling', () => {
   })
 })
 
+describe('getRetryAfterMs', () => {
+  it('finds a validated retry delay through an error cause chain', () => {
+    const providerError = Object.assign(new Error('rate limited'), { retryAfterMs: 45_000 })
+    expect(getRetryAfterMs(new Error('connector failed', { cause: providerError }))).toBe(45_000)
+  })
+
+  it.each([undefined, null, 0, -1, Number.NaN, Number.POSITIVE_INFINITY, '30000'])(
+    'ignores an invalid retry delay: %s',
+    (retryAfterMs) => {
+      expect(getRetryAfterMs(Object.assign(new Error('invalid'), { retryAfterMs }))).toBeUndefined()
+    }
+  )
+})
+
 describe('retryWithExponentialBackoff retry budget', () => {
   afterEach(() => {
     vi.useRealTimers()