mirror of
https://github.com/simstudioai/sim.git
synced 2026-09-21 13:00:04 +08:00
fix(knowledge): parse the stored artifact, not the document's display name (#6817)
* fix(knowledge): parse the stored artifact, not the document's display name A connector document's `filename` is a display name that 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, with `mimeType: 'text/plain'`. `processDocumentAsync` discards the processing filename the sync engine computes and rebuilds its input from the document row, so the parser was chosen from the display name and re-parsed extracted text as the source binary. In production that failed 1,379 SharePoint PDFs with `Invalid PDF structure.` and silently double-wrapped 364 spreadsheets — those reported `completed`, wrapping a second fake sheet around the connector's own extraction, because SheetJS accepts almost any input. Parser selection now prefers the extension of the object actually fetched, falling back to the filename/MIME path when the URL is not ours or the key carries no extension a parser claims. Both ingestion paths are honest under that rule because `fitStorageKeyName` preserves extensions through truncation: an upload keys on its original name, a connector document keys on what it stored. This layer is what covers the stuck-document retry sweep, which rebuilds its own input from the same display name — the sweep is the path that reprocesses the already-failed documents, so a fix confined to `processDocumentAsync` would have left the remediation itself broken. The defect predates the connectors that expose it: Box fetches Box-side text representations for `pdf`/`docx`/`xlsx` and stores them under the source name too, so it was latent there before SharePoint and OneDrive reached binary formats. `connectorArtifactFileName` now owns the `.txt` suffix that the parser choice depends on, so the invariant is structural instead of a convention repeated at four call sites per function. * fix(knowledge): raise the connector sync ceiling and tie it to the stale lock A 2,600-document library exhausted the 30-minute budget and the run was killed mid-listing, leaving the connector's `syncing` lock set until the scheduler reclaimed it. Raising the ceiling is not a lone constant, because reclaiming a stale lock flips the connector to `error` and frees it for another sync. A TTL at or below the run ceiling would hand the lock to a successor while the first sync is still writing — two syncs racing the same `(connectorId, externalId)` rows. The previous values, a 1800s run against a hard-coded 120-minute TTL declared in a different file, held that invariant only by coincidence. Both now derive from one another, with a test pinning the margin so the next raise cannot silently break it.
This commit is contained in:
@@ -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)
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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<typeof db, 'execute' | 'select'>
|
||||
|
||||
type DocOp =
|
||||
@@ -1657,17 +1671,17 @@ async function addDocument(
|
||||
): Promise<DocumentData> {
|
||||
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',
|
||||
|
||||
@@ -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)
|
||||
})
|
||||
})
|
||||
@@ -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
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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/<id>-Report.pdf`) and a connector document keys
|
||||
* on what it stored (`kb/<id>-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
|
||||
}
|
||||
|
||||
@@ -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')
|
||||
})
|
||||
})
|
||||
Reference in New Issue
Block a user