diff --git a/apps/sim/app/api/knowledge/connectors/sync/route.ts b/apps/sim/app/api/knowledge/connectors/sync/route.ts index fcec6ace21e..e6dd8177b1f 100644 --- a/apps/sim/app/api/knowledge/connectors/sync/route.ts +++ b/apps/sim/app/api/knowledge/connectors/sync/route.ts @@ -9,6 +9,7 @@ import { mapWithConcurrency } from '@/lib/core/utils/concurrency' import { generateRequestId } from '@/lib/core/utils/request' import { withRouteHandler } from '@/lib/core/utils/with-route-handler' import { dispatchSync } from '@/lib/knowledge/connectors/queue' +import { CONNECTOR_SYNC_STALE_LOCK_TTL_MS } from '@/lib/knowledge/connectors/sync-limits' export const dynamic = 'force-dynamic' @@ -39,8 +40,7 @@ export const GET = withRouteHandler(async (request: NextRequest) => { try { const now = new Date() - const STALE_SYNC_TTL_MS = 120 * 60 * 1000 - const staleCutoff = new Date(now.getTime() - STALE_SYNC_TTL_MS) + const staleCutoff = new Date(now.getTime() - CONNECTOR_SYNC_STALE_LOCK_TTL_MS) const recoveredConnectors = await db .update(knowledgeConnector) diff --git a/apps/sim/background/knowledge-connector-sync.ts b/apps/sim/background/knowledge-connector-sync.ts index ee17426f2bc..6efe6dbd26e 100644 --- a/apps/sim/background/knowledge-connector-sync.ts +++ b/apps/sim/background/knowledge-connector-sync.ts @@ -5,6 +5,7 @@ import { type ConnectorSyncPayload, } from '@/lib/knowledge/connectors/queue' import { executeSync } from '@/lib/knowledge/connectors/sync-engine' +import { CONNECTOR_SYNC_MAX_DURATION_SECONDS } from '@/lib/knowledge/connectors/sync-limits' const logger = createLogger('TriggerKnowledgeConnectorSync') @@ -39,7 +40,7 @@ export async function executeConnectorSyncJob(payload: unknown) { export const knowledgeConnectorSync = task({ id: 'knowledge-connector-sync', - maxDuration: 1800, + maxDuration: CONNECTOR_SYNC_MAX_DURATION_SECONDS, machine: 'large-2x', retry: { maxAttempts: 3, diff --git a/apps/sim/lib/knowledge/connectors/sync-engine.ts b/apps/sim/lib/knowledge/connectors/sync-engine.ts index ccd804a50df..dcbc354173b 100644 --- a/apps/sim/lib/knowledge/connectors/sync-engine.ts +++ b/apps/sim/lib/knowledge/connectors/sync-engine.ts @@ -64,6 +64,20 @@ const MAX_CONSECUTIVE_FAILURES = 10 function sanitizeStorageTitle(title: string): string { return title.replace(/[^a-zA-Z0-9.-]/g, '_').slice(0, MAX_SAFE_TITLE_LENGTH) } + +/** + * Name a connector document's stored object carries. + * + * Connectors store already-extracted text while `document.filename` keeps the + * source file's name for display, so the stored object has to declare the format + * it actually holds: `resolveStoredArtifactExtension` picks the parser off this + * key, and a key ending in the source extension would re-parse extracted text as + * the original binary. Owning the `.txt` suffix here makes that structural rather + * than a convention each call site has to remember. + */ +function connectorArtifactFileName(title: string): string { + return `${sanitizeStorageTitle(title)}.txt` +} type KnowledgeBaseLockingTx = Pick type DocOp = @@ -1657,17 +1671,17 @@ async function addDocument( ): Promise { const documentId = generateId() const contentBuffer = Buffer.from(extDoc.content, 'utf-8') - const safeTitle = sanitizeStorageTitle(extDoc.title) - const customKey = `kb/${buildStorageKeySegment(`${Date.now()}-${documentId}-`, `${safeTitle}.txt`)}` + const storedFileName = connectorArtifactFileName(extDoc.title) + const customKey = `kb/${buildStorageKeySegment(`${Date.now()}-${documentId}-`, storedFileName)}` const fileInfo = await StorageService.uploadFile({ file: contentBuffer, - fileName: `${safeTitle}.txt`, + fileName: storedFileName, contentType: 'text/plain', context: 'knowledge-base', customKey, preserveKey: true, - metadata: kbOwnershipMetadata(kbOwner, `${safeTitle}.txt`), + metadata: kbOwnershipMetadata(kbOwner, storedFileName), }) const fileUrl = `${getInternalApiBaseUrl()}${fileInfo.path}?context=knowledge-base` @@ -1676,8 +1690,6 @@ async function addDocument( ? resolveTagMapping(connectorType, extDoc.metadata, sourceConfig) : undefined - const processingFilename = `${safeTitle}.txt` - try { await db.transaction(async (tx) => { const isActive = await isKnowledgeBaseActiveInTx(tx, knowledgeBaseId) @@ -1718,7 +1730,7 @@ async function addDocument( return { documentId, - filename: processingFilename, + filename: storedFileName, fileUrl, fileSize: contentBuffer.length, mimeType: 'text/plain', @@ -1746,17 +1758,17 @@ async function updateDocument( const oldFileUrl = existingRows[0]?.fileUrl const contentBuffer = Buffer.from(extDoc.content, 'utf-8') - const safeTitle = sanitizeStorageTitle(extDoc.title) - const customKey = `kb/${buildStorageKeySegment(`${Date.now()}-${existingDocId}-`, `${safeTitle}.txt`)}` + const storedFileName = connectorArtifactFileName(extDoc.title) + const customKey = `kb/${buildStorageKeySegment(`${Date.now()}-${existingDocId}-`, storedFileName)}` const fileInfo = await StorageService.uploadFile({ file: contentBuffer, - fileName: `${safeTitle}.txt`, + fileName: storedFileName, contentType: 'text/plain', context: 'knowledge-base', customKey, preserveKey: true, - metadata: kbOwnershipMetadata(kbOwner, `${safeTitle}.txt`), + metadata: kbOwnershipMetadata(kbOwner, storedFileName), }) const fileUrl = `${getInternalApiBaseUrl()}${fileInfo.path}?context=knowledge-base` @@ -1765,8 +1777,6 @@ async function updateDocument( ? resolveTagMapping(connectorType, extDoc.metadata, sourceConfig) : undefined - const processingFilename = `${safeTitle}.txt` - try { await db.transaction(async (tx) => { const isActive = await isKnowledgeBaseActiveInTx(tx, knowledgeBaseId) @@ -1839,7 +1849,7 @@ async function updateDocument( return { documentId: existingDocId, - filename: processingFilename, + filename: storedFileName, fileUrl, fileSize: contentBuffer.length, mimeType: 'text/plain', diff --git a/apps/sim/lib/knowledge/connectors/sync-limits.test.ts b/apps/sim/lib/knowledge/connectors/sync-limits.test.ts new file mode 100644 index 00000000000..41f5a0da19e --- /dev/null +++ b/apps/sim/lib/knowledge/connectors/sync-limits.test.ts @@ -0,0 +1,26 @@ +/** + * @vitest-environment node + */ +import { describe, expect, it } from 'vitest' +import { + CONNECTOR_SYNC_MAX_DURATION_SECONDS, + CONNECTOR_SYNC_STALE_LOCK_TTL_MS, +} from '@/lib/knowledge/connectors/sync-limits' + +describe('connector sync limits', () => { + /** + * Reclaiming a stale lock frees it for another sync, so a TTL at or below the + * run ceiling would start a second sync while the first is still writing. This + * guards the invariant against a future hard-coded TTL, not the derivation. + */ + it('keeps at least a 2x margin between the run ceiling and the reclaim', () => { + expect(CONNECTOR_SYNC_STALE_LOCK_TTL_MS).toBeGreaterThanOrEqual( + CONNECTOR_SYNC_MAX_DURATION_SECONDS * 2 * 1000 + ) + }) + + /** A 2,600-document library exhausted the previous 1800s budget mid-listing. */ + it('allows a run longer than the half hour that timed out in production', () => { + expect(CONNECTOR_SYNC_MAX_DURATION_SECONDS).toBeGreaterThan(1800) + }) +}) diff --git a/apps/sim/lib/knowledge/connectors/sync-limits.ts b/apps/sim/lib/knowledge/connectors/sync-limits.ts new file mode 100644 index 00000000000..5752d05c6a1 --- /dev/null +++ b/apps/sim/lib/knowledge/connectors/sync-limits.ts @@ -0,0 +1,16 @@ +/** + * Wall-clock ceiling for a single connector sync run. A large document library + * needs more than the half hour this used to allow: a 2,600-document site + * exhausted the old budget and was killed mid-listing, leaving its `syncing` + * lock set until the scheduler reclaimed it. + */ +export const CONNECTOR_SYNC_MAX_DURATION_SECONDS = 3600 + +/** + * How long a connector may sit in `syncing` before the scheduler reclaims its lock. + * + * MUST stay above {@link CONNECTOR_SYNC_MAX_DURATION_SECONDS}: reclaiming frees the + * lock for another sync, so a TTL at or below the run ceiling would start a second + * sync while the first is still writing, both racing the same documents. + */ +export const CONNECTOR_SYNC_STALE_LOCK_TTL_MS = CONNECTOR_SYNC_MAX_DURATION_SECONDS * 2 * 1000 diff --git a/apps/sim/lib/knowledge/documents/document-processor.ts b/apps/sim/lib/knowledge/documents/document-processor.ts index 5ec1a39bda7..baa36447948 100644 --- a/apps/sim/lib/knowledge/documents/document-processor.ts +++ b/apps/sim/lib/knowledge/documents/document-processor.ts @@ -18,7 +18,10 @@ import { env, envNumber } from '@/lib/core/config/env' import { OCR_CAPABILITY, requireCapability } from '@/lib/core/config/env-capabilities' import { parseBuffer } from '@/lib/file-parsers' import type { FileParseMetadata } from '@/lib/file-parsers/types' -import { resolveParserExtension } from '@/lib/knowledge/documents/parser-extension' +import { + resolveParserExtension, + resolveStoredArtifactExtension, +} from '@/lib/knowledge/documents/parser-extension' import { retryWithExponentialBackoff } from '@/lib/knowledge/documents/utils' import { assertKnowledgeOpaqueModelInputSafe, @@ -841,7 +844,9 @@ async function parseHttpFile( ): Promise<{ content: string; metadata?: FileParseMetadata }> { const buffer = await downloadFileWithTimeout(fileUrl, userId) - const extension = resolveParserExtension(filename, mimeType) + /** Prefer what we actually downloaded over what the document is *called*. */ + const extension = + resolveStoredArtifactExtension(fileUrl) ?? resolveParserExtension(filename, mimeType) const result = await parseBuffer(buffer, extension) return result } diff --git a/apps/sim/lib/knowledge/documents/parser-extension.ts b/apps/sim/lib/knowledge/documents/parser-extension.ts index 39e7728b674..974b6235b1b 100644 --- a/apps/sim/lib/knowledge/documents/parser-extension.ts +++ b/apps/sim/lib/knowledge/documents/parser-extension.ts @@ -1,4 +1,9 @@ -import { getExtensionFromMimeType } from '@/lib/uploads/utils/file-utils' +import { + extractStorageKey, + getExtensionFromMimeType, + getFileExtension, + isInternalFileUrl, +} from '@/lib/uploads/utils/file-utils' import { isAlphanumericExtension, isSupportedExtension, @@ -12,8 +17,8 @@ export function resolveParserExtension( mimeType?: string, fallback?: string ): string { - const raw = filename.includes('.') ? filename.split('.').pop()?.toLowerCase() : undefined - const filenameExtension = raw && isAlphanumericExtension(raw) ? raw : undefined + const raw = getFileExtension(filename) + const filenameExtension = isAlphanumericExtension(raw) ? raw : undefined if (filenameExtension && isSupportedExtension(filenameExtension)) { return filenameExtension @@ -36,3 +41,31 @@ export function resolveParserExtension( throw new Error(`Could not determine file type for ${filename || 'document'}`) } + +/** + * Extension of the object actually stored, taken from its storage key. + * + * A knowledge base document's `filename` is a *display* name, and for connector + * documents it deliberately disagrees with the bytes on disk: the sync engine + * records the source file's name (`Report.pdf`) while storing the text the + * connector already extracted from it under a `.txt` key. Choosing a parser from + * the display name therefore re-parses extracted text as the original binary + * format — `Invalid PDF structure.` for PDFs, and for spreadsheets a silent + * double-wrap, since SheetJS accepts almost anything. + * + * The storage key is the honest signal for both ingestion paths, because + * `fitStorageKeyName` preserves a file's extension through truncation: an upload + * keys on its original name (`kb/-Report.pdf`) and a connector document keys + * on what it stored (`kb/-Report.pdf.txt`). + * + * Falls back to `undefined` — leaving the caller on the filename/MIME path — + * rather than guessing, so this can only ever redirect to a parser that exists. + */ +export function resolveStoredArtifactExtension(fileUrl: string): string | undefined { + if (!isInternalFileUrl(fileUrl)) return undefined + + const extension = getFileExtension(extractStorageKey(fileUrl)) + if (!isAlphanumericExtension(extension)) return undefined + + return isSupportedExtension(extension) ? extension : undefined +} diff --git a/apps/sim/lib/knowledge/documents/stored-artifact-extension.test.ts b/apps/sim/lib/knowledge/documents/stored-artifact-extension.test.ts new file mode 100644 index 00000000000..25760202244 --- /dev/null +++ b/apps/sim/lib/knowledge/documents/stored-artifact-extension.test.ts @@ -0,0 +1,73 @@ +/** + * @vitest-environment node + * + * A knowledge base document's `filename` is a display name. For connector + * documents it deliberately disagrees with the stored bytes — the sync engine + * records `Report.pdf` while storing the text the connector already extracted + * under a `.txt` key — so choosing a parser from the display name re-parsed + * extracted text as the source binary. In production that failed 1,379 + * SharePoint PDFs with `Invalid PDF structure.` and silently double-wrapped + * every spreadsheet, which "succeeded" because SheetJS accepts almost anything. + */ +import { describe, expect, it } from 'vitest' +import { resolveStoredArtifactExtension } from '@/lib/knowledge/documents/parser-extension' + +const CONNECTOR_PDF_URL = + '/api/files/serve/s3/kb%2F1786986883507-abc-Report.pdf.txt?context=knowledge-base' +const UPLOADED_PDF_URL = + '/api/files/serve/s3/kb%2F1786986883507-abc-Report.pdf?context=knowledge-base' + +describe('resolveStoredArtifactExtension', () => { + it('reports txt for a connector document whose display name is a PDF', () => { + expect(resolveStoredArtifactExtension(CONNECTOR_PDF_URL)).toBe('txt') + }) + + it('reports txt for a connector spreadsheet, which SheetJS would otherwise re-wrap', () => { + expect( + resolveStoredArtifactExtension( + '/api/files/serve/s3/kb%2F1-abc-Vendor_Spend.xlsx.txt?context=knowledge-base' + ) + ).toBe('txt') + }) + + it('leaves an uploaded document on its real extension', () => { + expect(resolveStoredArtifactExtension(UPLOADED_PDF_URL)).toBe('pdf') + }) + + it('handles the blob and gcs storage prefixes', () => { + expect(resolveStoredArtifactExtension('/api/files/serve/blob/kb%2F1-a-x.docx')).toBe('docx') + expect(resolveStoredArtifactExtension('/api/files/serve/gcs/kb%2F1-a-x.csv')).toBe('csv') + }) + + it('ignores URLs that are not served from our own storage', () => { + expect(resolveStoredArtifactExtension('https://example.com/files/Report.pdf')).toBeUndefined() + expect(resolveStoredArtifactExtension('data:application/pdf;base64,AAAA')).toBeUndefined() + }) + + /** + * `fitStorageKeyName` drops the extension when it cannot fit, and a key may + * carry no extension at all. Returning undefined puts the caller back on the + * filename/MIME path rather than guessing. + */ + it('returns undefined when the key carries no usable extension', () => { + expect(resolveStoredArtifactExtension('/api/files/serve/s3/kb%2F1-a-Report')).toBeUndefined() + expect(resolveStoredArtifactExtension('/api/files/serve/s3/kb%2F1-a-Report.')).toBeUndefined() + }) + + /** + * Only ever redirects to a parser that exists — an unknown suffix falls back + * instead of routing the document at a parser that cannot handle it. + */ + it('returns undefined for an extension no parser claims', () => { + expect( + resolveStoredArtifactExtension('/api/files/serve/s3/kb%2F1-a-archive.zip') + ).toBeUndefined() + expect( + resolveStoredArtifactExtension('/api/files/serve/s3/kb%2F1-a-Report.v2.final') + ).toBeUndefined() + }) + + it('is case-insensitive', () => { + expect(resolveStoredArtifactExtension('/api/files/serve/s3/kb%2F1-a-Report.PDF')).toBe('pdf') + }) +})