diff --git a/apps/sim/connectors/github/github.test.ts b/apps/sim/connectors/github/github.test.ts new file mode 100644 index 00000000000..c85e909206b --- /dev/null +++ b/apps/sim/connectors/github/github.test.ts @@ -0,0 +1,120 @@ +/** + * @vitest-environment node + */ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { githubConnector } from '@/connectors/github/github' + +describe('githubConnector.getDocument', () => { + afterEach(() => { + vi.unstubAllGlobals() + }) + + it('uses the object media type and hydrates large file content through the blob API', async () => { + const fetchMock = vi + .fn() + .mockResolvedValueOnce( + new Response( + JSON.stringify({ + sha: 'blob-sha', + size: 2 * 1024 * 1024, + content: '', + encoding: 'none', + }), + { status: 200, headers: { 'last-modified': 'Fri, 28 Aug 2026 12:00:00 GMT' } } + ) + ) + .mockResolvedValueOnce( + new Response('large text file', { status: 200, headers: { 'content-length': '15' } }) + ) + vi.stubGlobal('fetch', fetchMock) + + const document = await githubConnector.getDocument( + 'token', + { repository: 'owner/repo', branch: 'main' }, + 'docs/large.md' + ) + + expect(fetchMock).toHaveBeenCalledTimes(2) + expect(fetchMock.mock.calls[0][1]).toMatchObject({ + headers: expect.objectContaining({ Accept: 'application/vnd.github.object+json' }), + }) + expect(fetchMock.mock.calls[1][1]).toMatchObject({ + headers: expect.objectContaining({ Accept: 'application/vnd.github.raw+json' }), + }) + expect(document).toMatchObject({ + externalId: 'docs/large.md', + content: 'large text file', + contentDeferred: false, + contentHash: 'git-sha:blob-sha', + }) + }) + + it('returns null only when a listed path is no longer present', async () => { + vi.stubGlobal('fetch', vi.fn().mockResolvedValue(new Response(null, { status: 404 }))) + + await expect( + githubConnector.getDocument('token', { repository: 'owner/repo' }, 'deleted.md') + ).resolves.toBeNull() + }) + + it('records a blob that exceeds the byte cap as a visible skipped document', async () => { + const fetchMock = vi + .fn() + .mockResolvedValueOnce( + new Response( + JSON.stringify({ + sha: 'blob-sha', + size: 2 * 1024 * 1024, + content: '', + encoding: 'none', + }), + { status: 200 } + ) + ) + .mockResolvedValueOnce( + new Response('oversized', { + status: 200, + headers: { 'content-length': String(100 * 1024 * 1024 + 1) }, + }) + ) + vi.stubGlobal('fetch', fetchMock) + + await expect( + githubConnector.getDocument('token', { repository: 'owner/repo' }, 'oversized.md') + ).resolves.toMatchObject({ + externalId: 'oversized.md', + content: '', + skippedReason: 'File exceeds the 100MB size limit and was not indexed', + }) + }) + + it('rejects a bodyless blob response instead of misreporting it as oversized', async () => { + const fetchMock = vi + .fn() + .mockResolvedValueOnce( + new Response( + JSON.stringify({ + sha: 'blob-sha', + size: 2 * 1024 * 1024, + content: '', + encoding: 'none', + }), + { status: 200 } + ) + ) + .mockResolvedValueOnce(new Response(null, { status: 200 })) + vi.stubGlobal('fetch', fetchMock) + + await expect( + githubConnector.getDocument('token', { repository: 'owner/repo' }, 'missing-body.md') + ).rejects.toThrow('GitHub git blob blob-sha returned no body') + }) + + it('surfaces a non-rate-limit 403 as a document failure', async () => { + vi.stubGlobal('fetch', vi.fn().mockResolvedValue(new Response(null, { status: 403 }))) + + await expect( + githubConnector.getDocument('token', { repository: 'owner/repo' }, 'private.md') + ).rejects.toThrow('Failed to fetch file private.md: 403') + }) +}) diff --git a/apps/sim/connectors/github/github.ts b/apps/sim/connectors/github/github.ts index b1d1356354f..e26dc938e64 100644 --- a/apps/sim/connectors/github/github.ts +++ b/apps/sim/connectors/github/github.ts @@ -5,8 +5,10 @@ import { githubConnectorMeta } from '@/connectors/github/meta' import type { ConnectorConfig, ExternalDocument, ExternalDocumentList } from '@/connectors/types' import { CONNECTOR_MAX_FILE_BYTES, + ConnectorFileTooLargeError, markSkipped, parseTagDate, + readBodyWithLimit, sizeLimitSkipReason, stubOrSkipBySize, takeIndexableWithinCap, @@ -156,7 +158,7 @@ async function fetchBlobContent( const response = await fetchWithRetry(url, { method: 'GET', headers: { - Accept: 'application/vnd.github+json', + Accept: 'application/vnd.github.raw+json', Authorization: `Bearer ${accessToken}`, 'X-GitHub-Api-Version': '2022-11-28', }, @@ -166,25 +168,20 @@ async function fetchBlobContent( throw new Error(`Failed to fetch git blob ${sha}: ${response.status}`) } - const data = await response.json() - const content = (data.content as string) || '' - const encoding = data.encoding as string | undefined + if (!response.body) { + const contentLength = Number.parseInt(response.headers.get('content-length') ?? '', 10) + if (Number.isFinite(contentLength) && contentLength > MAX_FILE_SIZE) { + throw new ConnectorFileTooLargeError(MAX_FILE_SIZE) + } + throw new Error(`GitHub git blob ${sha} returned no body`) + } - if (encoding === 'base64') { - const buf = Buffer.from(content, 'base64') - if (isBinaryBuffer(buf)) return null - return buf.toString('utf8') + const buffer = await readBodyWithLimit(response, MAX_FILE_SIZE) + if (!buffer) { + throw new ConnectorFileTooLargeError(MAX_FILE_SIZE) } - /** - * `GET /repos/{owner}/{repo}/git/blobs/{sha}` documents a single response - * encoding: "The `content` in the response will always be Base64 encoded." - * The "Currently, `utf-8` and `base64` are supported" sentence belongs to the - * `encoding` REQUEST parameter of `POST .../git/blobs` (Create a blob) and does - * not describe this response, so no `utf-8` branch is warranted here. Any other - * encoding would silently persist empty content, so it throws and surfaces as a - * failed document instead. - */ - throw new Error(`Unexpected git blob encoding for ${sha}: ${encoding ?? 'undefined'}`) + if (isBinaryBuffer(buffer)) return null + return buffer.toString('utf8') } /** @@ -325,7 +322,7 @@ export const githubConnector: ConnectorConfig = { const response = await fetchWithRetry(url, { method: 'GET', headers: { - Accept: 'application/vnd.github+json', + Accept: 'application/vnd.github.object+json', Authorization: `Bearer ${accessToken}`, 'X-GitHub-Api-Version': '2022-11-28', }, @@ -333,25 +330,6 @@ export const githubConnector: ConnectorConfig = { if (!response.ok) { if (response.status === 404) return null - /** - * A rate-limit 403 never reaches here: `fetchWithRetry` treats a 403 carrying - * `retry-after` or `x-ratelimit-remaining: 0` as retryable and throws once the - * retries are spent, so it lands in the catch below as a failure. - * - * A 403 that survives is usually an authorization denial, but NOT always: this - * request sends `application/vnd.github+json`, and the Contents API documents - * that files between 1-100 MB support "only the `raw` or `object` custom media - * types". A >1 MB text file therefore also lands here and is dropped, which on - * an `add` is silent (a fulfilled `null` records no failure). Reconciliation is - * unaffected — the file is already in `seenExternalIds` from the listing. - */ - if (response.status === 403) { - logger.info('Skipping GitHub file rejected by Contents API', { - path, - status: response.status, - }) - return null - } throw new Error(`Failed to fetch file ${path}: ${response.status}`) } @@ -392,15 +370,18 @@ export const githubConnector: ConnectorConfig = { * "only the `raw` or `object` custom media types are supported", and it is * specifically "when using the `object` media type" that "the `content` field * will be an empty string and the `encoding` field will be `none`". - * - * This request sends `application/vnd.github+json`, so that precondition does - * not hold and this branch is currently unreachable — such files 403 above - * instead. Reaching it would require requesting - * `application/vnd.github.object+json`. The fallback itself is correct: the Git - * Blobs API returns the same blob as JSON and is documented to support blobs up - * to 100 MB. + * The Git Blobs fallback streams that same blob through GitHub's documented raw + * media type and supports blobs up to 100 MB. */ - const blobContent = await fetchBlobContent(accessToken, owner, repo, data.sha as string) + let blobContent: string | null + try { + blobContent = await fetchBlobContent(accessToken, owner, repo, data.sha as string) + } catch (error) { + if (error instanceof ConnectorFileTooLargeError) { + return markSkipped(stub, sizeLimitSkipReason(MAX_FILE_SIZE)) + } + throw error + } if (blobContent === null) { logger.info('Skipping binary GitHub file', { path, size }) return markSkipped(stub, BINARY_SKIP_REASON) diff --git a/apps/sim/connectors/google-drive/google-drive-errors.ts b/apps/sim/connectors/google-drive/google-drive-errors.ts index fb2ffeb81e1..76f9449542a 100644 --- a/apps/sim/connectors/google-drive/google-drive-errors.ts +++ b/apps/sim/connectors/google-drive/google-drive-errors.ts @@ -12,13 +12,12 @@ const PERMISSION_REASONS = new Set([ const POLICY_REASONS = new Set(['domainPolicy', 'download_restricted_for_revision']) const UNSUPPORTED_EXPORT_REASONS = new Set(['fileNotDownloadable', 'fileNotExportable']) const QUOTA_REASONS = new Set(['dailyLimitExceeded', 'quotaExceeded']) -const TRANSIENT_REASONS = new Set([ - 'backendError', - 'internalError', +const RATE_LIMIT_REASONS = new Set([ 'rateLimitExceeded', 'sharingRateLimitExceeded', 'userRateLimitExceeded', ]) +const TRANSIENT_REASONS = new Set(['backendError', 'internalError', ...RATE_LIMIT_REASONS]) export type GoogleDriveErrorKind = | 'authorization' @@ -99,15 +98,22 @@ function classifyGoogleDriveError( export class GoogleDriveApiError extends Error { retryAfterMs?: number + readonly reasons: readonly string[] + readonly kind: GoogleDriveErrorKind + readonly rateLimited: boolean constructor( readonly status: number, - readonly reasons: readonly string[], - readonly kind: GoogleDriveErrorKind + normalizedReasons: readonly string[] ) { - const reasonSuffix = reasons.length > 0 ? ` (${reasons.join(', ')})` : '' + const diagnosticReasons = normalizedReasons.slice(0, GOOGLE_ERROR_REASON_MAX_COUNT) + const reasonSuffix = diagnosticReasons.length > 0 ? ` (${diagnosticReasons.join(', ')})` : '' super(`Google Drive API request failed with HTTP ${status}${reasonSuffix}`) this.name = 'GoogleDriveApiError' + this.reasons = diagnosticReasons + this.kind = classifyGoogleDriveError(status, normalizedReasons) + this.rateLimited = + status === 429 || normalizedReasons.some((reason) => RATE_LIMIT_REASONS.has(reason)) } } @@ -130,13 +136,8 @@ export async function readGoogleDriveApiError(response: Response): Promise (entry.reason ? [entry.reason] : [])))] - const reasons = [...new Set(rawReasons.flatMap((reason) => normalizeReason(reason) ?? []))].slice( - 0, - GOOGLE_ERROR_REASON_MAX_COUNT - ) - return new GoogleDriveApiError( - response.status, - reasons, - classifyGoogleDriveError(response.status, rawReasons) - ) + const normalizedReasons = [ + ...new Set(rawReasons.flatMap((reason) => normalizeReason(reason) ?? [])), + ] + return new GoogleDriveApiError(response.status, normalizedReasons) } diff --git a/apps/sim/connectors/google-drive/google-drive.test.ts b/apps/sim/connectors/google-drive/google-drive.test.ts index fde940e924f..3db406ff869 100644 --- a/apps/sim/connectors/google-drive/google-drive.test.ts +++ b/apps/sim/connectors/google-drive/google-drive.test.ts @@ -94,6 +94,39 @@ describe('Google Drive API error parsing', () => { expect(error.message).not.toContain('upstream unavailable') }) + it('normalizes only structured rate-limit reasons into the shared throttle signal', async () => { + const rateLimit = await readGoogleDriveApiError( + driveErrorResponse('userRateLimitExceeded', 'Provider message') + ) + const backendFailure = await readGoogleDriveApiError( + driveErrorResponse('backendError', 'Provider message', 503) + ) + + expect(rateLimit.rateLimited).toBe(true) + expect(backendFailure.rateLimited).toBe(false) + }) + + it('detects a structured rate limit beyond the bounded diagnostic reasons', async () => { + const reasons = [ + ...Array.from({ length: 16 }, (_, index) => `providerReason${index}`), + 'userRateLimitExceeded', + ] + const error = await readGoogleDriveApiError( + jsonResponse( + { + error: { + errors: reasons.map((reason) => ({ reason })), + }, + }, + 403 + ) + ) + + expect(error.reasons).toEqual(reasons.slice(0, 16)) + expect(error.kind).toBe('transient') + expect(error.rateLimited).toBe(true) + }) + it('omits provider messages from diagnostics', async () => { const message = `Authorization: Bearer private-token\ncontext ${'x'.repeat(700)}` const error = await readGoogleDriveApiError( diff --git a/apps/sim/lib/knowledge/connectors/sync-engine.test.ts b/apps/sim/lib/knowledge/connectors/sync-engine.test.ts index 7ce9da29933..12a5919fafd 100644 --- a/apps/sim/lib/knowledge/connectors/sync-engine.test.ts +++ b/apps/sim/lib/knowledge/connectors/sync-engine.test.ts @@ -51,7 +51,8 @@ vi.mock('@/background/knowledge-connector-sync', () => ({ knowledgeConnectorSync: { trigger: vi.fn() }, })) -const { mockMapTags, mockListDocuments } = vi.hoisted(() => ({ +const { mockGetDocument, mockMapTags, mockListDocuments } = vi.hoisted(() => ({ + mockGetDocument: vi.fn(), mockMapTags: vi.fn(), mockListDocuments: vi.fn(), })) @@ -71,6 +72,7 @@ vi.mock('@/connectors/registry.server', () => ({ paged: { name: 'Paged', auth: { mode: 'apiKey', optional: true }, + getDocument: mockGetDocument, listDocuments: mockListDocuments, }, }, @@ -1166,6 +1168,101 @@ describe('executeSync working-set overflow admission', () => { ) }) +describe('executeSync deferred hydration rate limits', () => { + const NOW = new Date('2026-08-29T03:00:00.000Z') + const CONNECTOR = { + id: 'c-1', + knowledgeBaseId: 'kb-1', + connectorType: 'paged', + credentialId: null, + encryptedApiKey: null, + sourceConfig: {}, + syncMode: 'full', + syncIntervalMinutes: 1440, + status: 'active', + lastSyncAt: null, + lastSyncDocCount: null, + consecutiveFailures: 0, + syncLockToken: null, + } + + const deferredDocument = (index: number): ExternalDocument => ({ + externalId: `external-${index}`, + title: `Document ${index}`, + content: '', + contentDeferred: true, + contentHash: `hash-${index}`, + mimeType: 'text/plain', + metadata: { size: 1024 }, + }) + + beforeEach(() => { + vi.clearAllMocks() + resetDbChainMock() + vi.useFakeTimers() + vi.setSystemTime(NOW) + queueTableRows(schemaMock.knowledgeConnector, [CONNECTOR]) + queueTableRows(schemaMock.knowledgeBase, [{ userId: 'u-1', workspaceId: 'ws-1' }]) + queueTableRows(schemaMock.document, []) + queueTableRows(schemaMock.document, []) + queueTableRows(schemaMock.document, []) + queueTableRows(schemaMock.document, []) + queueTableRows(schemaMock.knowledgeConnector, [ + { connectorArchivedAt: null, connectorDeletedAt: null, kbDeletedAt: null }, + ]) + dbChainMockFns.returning.mockResolvedValueOnce([CONNECTOR]) + }) + + afterEach(() => { + vi.useRealTimers() + }) + + it('stops after the active batch and preserves the provider retry delay', async () => { + const { executeSync } = await import('@/lib/knowledge/connectors/sync-engine') + const documents = Array.from({ length: 6 }, (_, index) => deferredDocument(index)) + const rateLimitError = Object.assign(new Error('HTTP 403 - upstream rate limit exceeded'), { + status: 403, + headers: new Headers({ 'x-ratelimit-remaining': '0' }), + retryAfterMs: 45 * 60 * 1000, + }) + + mockListDocuments.mockResolvedValue({ documents, hasMore: false }) + mockGetDocument.mockImplementation(async (_token, _config, externalId: string) => { + if (externalId === 'external-2') throw rateLimitError + return { + ...documents[Number(externalId.slice('external-'.length))], + content: 'hydrated', + contentDeferred: false, + } + }) + + const result = await executeSync('c-1', { + billingAttribution: { workspaceId: 'ws-1' } as never, + }) + + expect(mockGetDocument).toHaveBeenCalledTimes(5) + expect(mockGetDocument).not.toHaveBeenCalledWith( + expect.anything(), + expect.anything(), + 'external-5', + expect.anything() + ) + expect(result).toMatchObject({ + docsAdded: 0, + docsFailed: 0, + error: rateLimitError.message, + }) + expect(mockUploadFile).not.toHaveBeenCalled() + expect(mockProcessDocumentsWithQueue).not.toHaveBeenCalled() + expect(dbChainMockFns.set).toHaveBeenCalledWith( + expect.objectContaining({ + status: 'error', + nextSyncAt: new Date(NOW.getTime() + 45 * 60 * 1000), + }) + ) + }) +}) + describe('classifySuspectListing', () => { it('trusts a healthy listing', () => { expect(classifySuspectListing(100, 100)).toBeNull() diff --git a/apps/sim/lib/knowledge/connectors/sync-engine.ts b/apps/sim/lib/knowledge/connectors/sync-engine.ts index c57713c31e8..7c7b8d059a1 100644 --- a/apps/sim/lib/knowledge/connectors/sync-engine.ts +++ b/apps/sim/lib/knowledge/connectors/sync-engine.ts @@ -53,7 +53,7 @@ import { MAX_PROCESSING_ATTEMPTS, QUEUED_DISPATCH_GRACE_MS, } from '@/lib/knowledge/documents/types' -import { getRetryAfterMs } from '@/lib/knowledge/documents/utils' +import { getRetryAfterMs, isRateLimitError } 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' @@ -2443,6 +2443,14 @@ export async function executeSync( }) ) + const rateLimitFailure = hydrated.find( + (outcome): outcome is PromiseRejectedResult => + outcome.status === 'rejected' && isRateLimitError(outcome.reason) + ) + if (rateLimitFailure) { + throw rateLimitFailure.reason + } + for (let i = 0; i < hydrated.length; i++) { const outcome = hydrated[i] if (outcome.status === 'fulfilled' && outcome.value) { diff --git a/apps/sim/lib/knowledge/documents/utils.test.ts b/apps/sim/lib/knowledge/documents/utils.test.ts index 4f8c307a4e1..03cf245c8c1 100644 --- a/apps/sim/lib/knowledge/documents/utils.test.ts +++ b/apps/sim/lib/knowledge/documents/utils.test.ts @@ -17,6 +17,7 @@ import { getRetryAfterMs, type HTTPError, hasRateLimitEvidence, + isRateLimitError, isRetryableError, readBoundedHttpErrorPayload, resolveRetryDelayMs, @@ -640,6 +641,45 @@ describe('getRetryAfterMs', () => { ) }) +describe('isRateLimitError', () => { + it('accepts a 429 without requiring response headers', () => { + expect(isRateLimitError(Object.assign(new Error('throttled'), { status: 429 }))).toBe(true) + }) + + it('accepts a GitHub 403 only with structured rate-limit evidence', () => { + expect( + isRateLimitError( + Object.assign(new Error('forbidden'), { + status: 403, + headers: headers({ 'x-ratelimit-remaining': '0' }), + }) + ) + ).toBe(true) + expect(isRateLimitError(Object.assign(new Error('forbidden'), { status: 403 }))).toBe(false) + }) + + it('finds a structured rate-limit rejection through an error cause chain', () => { + const providerError = Object.assign(new Error('throttled'), { status: 429 }) + expect(isRateLimitError(new Error('hydration failed', { cause: providerError }))).toBe(true) + }) + + it('accepts a provider-normalized throttle without HTTP rate-limit headers', () => { + expect( + isRateLimitError( + Object.assign(new Error('Google Drive API request failed with HTTP 403'), { + status: 403, + rateLimited: true, + }) + ) + ).toBe(true) + }) + + it('does not classify retryable text or transient HTTP failures as provider throttling', () => { + expect(isRateLimitError(new Error('rate limit exceeded'))).toBe(false) + expect(isRateLimitError(Object.assign(new Error('unavailable'), { status: 503 }))).toBe(false) + }) +}) + 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 dc61d725553..d4357098ac9 100644 --- a/apps/sim/lib/knowledge/documents/utils.ts +++ b/apps/sim/lib/knowledge/documents/utils.ts @@ -26,6 +26,8 @@ export interface HTTPError extends Error { status?: number statusText?: string retryAfterMs?: number + /** Provider-normalized signal for throttles that do not carry standard HTTP evidence. */ + rateLimited?: boolean /** * Response headers carried onto the error so the retry loop can re-evaluate * rate-limit evidence (`isRetryableError` runs again on the thrown error). @@ -227,6 +229,32 @@ export function hasRateLimitEvidence(headers: HeaderReader | undefined): boolean return RATE_LIMIT_REMAINING_HEADERS.some((name) => headers.get(name) === '0') } +/** + * Reports whether an error or one of its causes is a structured HTTP rate-limit + * rejection. A bare 403 is intentionally excluded because it normally means the + * caller lacks access; GitHub identifies the rate-limit form through response + * headers, while 429 is unambiguous on its own. + */ +export function isRateLimitError(error: unknown): boolean { + const seen = new Set() + let current = error + + while (isRetryableErrorType(current) && !seen.has(current) && seen.size < 10) { + seen.add(current) + if ((current as HTTPError).rateLimited === true) return true + if ( + hasStatus(current) && + (current.status === 429 || + (current.status === 403 && hasRateLimitEvidence(readHeaders(current)))) + ) { + return true + } + current = current instanceof Error ? current.cause : undefined + } + + return false +} + function parseRateLimitResetMs(value: string, nowMs: number): number | undefined { const resetEpochSeconds = Number(value) if (!Number.isFinite(resetEpochSeconds) || resetEpochSeconds <= 0) return undefined