fix(connectors): honor provider retry deadlines

Author: waleedlatif1

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

@@ -205,6 +205,16 @@ 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'
+    )
   })
 })

apps/sim/lib/atlassian/discovery.ts

@@ -203,8 +203,15 @@ 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)) {
+    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)

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,10 +536,14 @@ 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)
   })
 
@@ -601,6 +606,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()

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

@@ -183,6 +183,30 @@ export function attachRetryHeaders(error: HTTPError, headers: HeaderReader): voi
   })
 }
 
+/**
+ * Reads a validated provider retry delay from an error or one of its causes.
+ *
+ * The HTTP retry layer attaches this value when a provider supplies
+ * `Retry-After` or an exhausted-quota reset header. Keeping the accessor here
+ * lets longer-lived schedulers honor the same evidence without depending on a
+ * concrete error class or parsing a diagnostic message.
+ */
+export function getRetryAfterMs(error: unknown): number | undefined {
+  const seen = new Set<unknown>()
+  let current = error
+
+  while (current instanceof Error && !seen.has(current) && seen.size < 10) {
+    seen.add(current)
+    const retryAfterMs = (current as HTTPError).retryAfterMs
+    if (typeof retryAfterMs === 'number' && Number.isFinite(retryAfterMs) && retryAfterMs > 0) {
+      return retryAfterMs
+    }
+    current = current.cause
+  }
+
+  return undefined
+}
+
 /**
  * True when response headers positively identify a rate-limit rejection rather
  * than an authorization denial.
@@ -254,6 +278,39 @@ export function resolveRetryDelayMs(
   return undefined
 }
 
+interface RetryableHttpResponse {
+  status: number
+  headers: { get(name: string): string | null }
+  body?: ReadableStream<Uint8Array> | null
+  arrayBuffer?: () => Promise<ArrayBuffer>
+  text?: () => Promise<string>
+}
+
+/**
+ * Builds the bounded error shared by direct and SSRF-safe connector fetches.
+ * Rate-limit responses are named from trusted status/header evidence while all
+ * provider-controlled bodies remain omitted.
+ */
+export async function createRetryableHttpError(
+  response: RetryableHttpResponse
+): Promise<HTTPError> {
+  const rateLimited =
+    response.status === 429 || (response.status === 403 && hasRateLimitEvidence(response.headers))
+  const diagnostic = rateLimited
+    ? 'upstream rate limit exceeded'
+    : await readBoundedHttpErrorBody(response)
+  const error: HTTPError = new Error(`HTTP ${response.status} - ${diagnostic}`)
+  error.status = response.status
+  attachRetryHeaders(error, response.headers)
+
+  const waitMs = resolveRetryDelayMs(response.headers)
+  if (waitMs !== undefined) {
+    error.retryAfterMs = waitMs
+  }
+
+  return error
+}
+
 /**
  * Default retry condition for rate limiting errors
  */
@@ -471,22 +528,7 @@ export async function fetchWithRetry(
     const response = await fetch(url, options)
 
     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
-      // The retry loop re-runs the retry condition against this error, so the
-      // headers must travel with it or a rate-limit 403 would throw immediately.
-      attachRetryHeaders(error, response.headers)
-
-      // Pass the server-stated wait to the retry loop so it replaces exponential
-      // backoff. Falls back to the epoch-seconds reset header when the provider
-      // sends no Retry-After (X never does).
-      const waitMs = resolveRetryDelayMs(response.headers)
-      if (waitMs !== undefined) {
-        error.retryAfterMs = waitMs
-      }
-
-      throw error
+      throw await createRetryableHttpError(response)
     }
 
     return response