From fd270d86b8b4777caa005157781090c9c81d25c7 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 28 Aug 2026 01:54:11 -0700 Subject: [PATCH 1/2] fix(connectors): honor provider retry deadlines --- apps/sim/lib/atlassian/discovery.test.ts | 12 ++- apps/sim/lib/atlassian/discovery.ts | 11 ++- .../knowledge/connectors/sync-engine.test.ts | 24 ++++++ .../lib/knowledge/connectors/sync-engine.ts | 32 ++++++-- .../documents/secure-fetch.server.ts | 17 +---- .../sim/lib/knowledge/documents/utils.test.ts | 23 +++++- apps/sim/lib/knowledge/documents/utils.ts | 74 +++++++++++++++---- 7 files changed, 152 insertions(+), 41 deletions(-) diff --git a/apps/sim/lib/atlassian/discovery.test.ts b/apps/sim/lib/atlassian/discovery.test.ts index d3c3f4b34bc..98490d18276 100644 --- a/apps/sim/lib/atlassian/discovery.test.ts +++ b/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' + ) }) }) diff --git a/apps/sim/lib/atlassian/discovery.ts b/apps/sim/lib/atlassian/discovery.ts index 5a27b134dc2..8b4581b259f 100644 --- a/apps/sim/lib/atlassian/discovery.ts +++ b/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) diff --git a/apps/sim/lib/knowledge/connectors/sync-engine.test.ts b/apps/sim/lib/knowledge/connectors/sync-engine.test.ts index dbb58359347..7ce9da29933 100644 --- a/apps/sim/lib/knowledge/connectors/sync-engine.test.ts +++ b/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') diff --git a/apps/sim/lib/knowledge/connectors/sync-engine.ts b/apps/sim/lib/knowledge/connectors/sync-engine.ts index bdca8b716f0..c57713c31e8 100644 --- a/apps/sim/lib/knowledge/connectors/sync-engine.ts +++ b/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,7 +3172,12 @@ 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 }) @@ -3168,7 +3185,12 @@ export async function executeSync( 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', { diff --git a/apps/sim/lib/knowledge/documents/secure-fetch.server.ts b/apps/sim/lib/knowledge/documents/secure-fetch.server.ts index 8c40b863542..818174760af 100644 --- a/apps/sim/lib/knowledge/documents/secure-fetch.server.ts +++ b/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 diff --git a/apps/sim/lib/knowledge/documents/utils.test.ts b/apps/sim/lib/knowledge/documents/utils.test.ts index 260743cf5d9..aa523776d64 100644 --- a/apps/sim/lib/knowledge/documents/utils.test.ts +++ b/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() diff --git a/apps/sim/lib/knowledge/documents/utils.ts b/apps/sim/lib/knowledge/documents/utils.ts index 5a54d0d7204..77d9acd0a90 100644 --- a/apps/sim/lib/knowledge/documents/utils.ts +++ b/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() + 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 | null + arrayBuffer?: () => Promise + text?: () => Promise +} + +/** + * 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 { + 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 From 24ec909d2e2049043d9c69b7557f81fc940e1075 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 28 Aug 2026 02:09:45 -0700 Subject: [PATCH 2/2] fix(connectors): validate retry response lifecycles --- apps/sim/lib/atlassian/discovery.test.ts | 14 +++++++++++++ apps/sim/lib/atlassian/discovery.ts | 21 +++++++++++++------ .../sim/lib/knowledge/documents/utils.test.ts | 20 ++++++++++++++++++ apps/sim/lib/knowledge/documents/utils.ts | 13 ++++++++++++ 4 files changed, 62 insertions(+), 6 deletions(-) diff --git a/apps/sim/lib/atlassian/discovery.test.ts b/apps/sim/lib/atlassian/discovery.test.ts index 98490d18276..d839c95f662 100644 --- a/apps/sim/lib/atlassian/discovery.test.ts +++ b/apps/sim/lib/atlassian/discovery.test.ts @@ -217,4 +217,18 @@ describe('resolveAtlassianCloudId', () => { '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' + ) + }) }) diff --git a/apps/sim/lib/atlassian/discovery.ts b/apps/sim/lib/atlassian/discovery.ts index 8b4581b259f..c1a55718807 100644 --- a/apps/sim/lib/atlassian/discovery.ts +++ b/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 + 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,7 +214,7 @@ export function selectAtlassianCloudId( domain: string, product: string ): string { - if (!Array.isArray(resources)) { + if (!Array.isArray(resources) || !resources.every(isAccessibleResource)) { throw new Error(`Invalid ${product} accessible-resources response`) } @@ -215,16 +226,14 @@ export function selectAtlassianCloudId( } 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(', ')}` ) } diff --git a/apps/sim/lib/knowledge/documents/utils.test.ts b/apps/sim/lib/knowledge/documents/utils.test.ts index aa523776d64..4f8c307a4e1 100644 --- a/apps/sim/lib/knowledge/documents/utils.test.ts +++ b/apps/sim/lib/knowledge/documents/utils.test.ts @@ -547,6 +547,26 @@ describe('fetchWithRetry rate-limit handling', () => { expect(fetchMock).toHaveBeenCalledTimes(1) }) + it('cancels an omitted rate-limit response body before throwing', async () => { + let cancelled = false + const body = new ReadableStream({ + 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 diff --git a/apps/sim/lib/knowledge/documents/utils.ts b/apps/sim/lib/knowledge/documents/utils.ts index 77d9acd0a90..dc61d725553 100644 --- a/apps/sim/lib/knowledge/documents/utils.ts +++ b/apps/sim/lib/knowledge/documents/utils.ts @@ -286,6 +286,16 @@ interface RetryableHttpResponse { text?: () => Promise } +/** Releases a response stream when its provider-controlled body is intentionally omitted. */ +async function cancelHttpResponseBody(response: RetryableHttpResponse): Promise { + if (!response.body) return + try { + await response.body.cancel() + } catch { + return + } +} + /** * Builds the bounded error shared by direct and SSRF-safe connector fetches. * Rate-limit responses are named from trusted status/header evidence while all @@ -296,6 +306,9 @@ export async function createRetryableHttpError( ): Promise { const rateLimited = response.status === 429 || (response.status === 403 && hasRateLimitEvidence(response.headers)) + if (rateLimited) { + await cancelHttpResponseBody(response) + } const diagnostic = rateLimited ? 'upstream rate limit exceeded' : await readBoundedHttpErrorBody(response)