Skip to content

Commit 95ad337

Browse files
waleedlatif1waleedlatif1
committed
fix(knowledge): enforce complete ingestion boundaries
1 parent 9757c98 commit 95ad337

33 files changed

Lines changed: 562 additions & 1405 deletions

‎apps/sim/background/knowledge-processing.test.ts‎

Lines changed: 40 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@ vi.mock('@/lib/knowledge/documents/service', () => ({
2929
import { EmbeddingQuotaExhaustedError } from '@/lib/embeddings/client'
3030
import { EMBEDDING_QUOTA_CIRCUIT_TTL_MS } from '@/lib/embeddings/quota-circuit'
3131
import { PermanentDocumentProcessingError } from '@/lib/knowledge/documents/document-processing-error'
32+
import { MAX_QUOTA_CONTINUATION_ATTEMPTS } from '@/lib/knowledge/documents/processing-quota-continuation'
3233
import {
3334
resolveQuotaContinuationDelayMs,
3435
runDocumentProcessing,
@@ -345,11 +346,39 @@ describe('knowledge processing worker', () => {
345346
quotaRetryCount: 1,
346347
}),
347348
expect.objectContaining({
348-
delay: new Date(1_000 + EMBEDDING_QUOTA_CIRCUIT_TTL_MS),
349+
delay: expect.any(Date),
349350
idempotencyKey: 'knowledge-quota-document-1-request-1-1',
350351
region: 'us-east-1',
351352
})
352353
)
354+
const delay = mockTrigger.mock.calls[0]?.[2]?.delay as Date
355+
expect(delay.getTime()).toBeGreaterThanOrEqual(1_000 + EMBEDDING_QUOTA_CIRCUIT_TTL_MS * 0.8)
356+
expect(delay.getTime()).toBeLessThanOrEqual(1_000 + EMBEDDING_QUOTA_CIRCUIT_TTL_MS * 1.2)
357+
})
358+
359+
it('ends a quota chain after the bounded continuation horizon', async () => {
360+
mockQuotaExhaustion(new EmbeddingQuotaExhaustedError('openai'))
361+
362+
await expect(
363+
runDocumentProcessing({
364+
...BASE_PAYLOAD,
365+
quotaRetryCount: MAX_QUOTA_CONTINUATION_ATTEMPTS,
366+
billingScope: 'non-workspace',
367+
actorUserId: 'legacy-owner',
368+
workspaceId: null,
369+
})
370+
).resolves.toMatchObject({ success: false, outcome: 'quota_exhausted' })
371+
372+
expect(mockTrigger).not.toHaveBeenCalled()
373+
expect(mockProcessDocumentAsync).toHaveBeenCalledWith(
374+
expect.any(String),
375+
expect.any(String),
376+
expect.any(Object),
377+
expect.any(Object),
378+
expect.any(Object),
379+
expect.any(String),
380+
expect.objectContaining({ quotaContinuationExhausted: true })
381+
)
353382
})
354383

355384
it('keeps the task failed when the durable continuation handoff fails', async () => {
@@ -456,8 +485,15 @@ describe('knowledge-process-document task configuration', () => {
456485
})
457486

458487
it('backs durable quota continuations off to a bounded polling interval', () => {
459-
expect(resolveQuotaContinuationDelayMs(1)).toBe(EMBEDDING_QUOTA_CIRCUIT_TTL_MS)
460-
expect(resolveQuotaContinuationDelayMs(2)).toBe(EMBEDDING_QUOTA_CIRCUIT_TTL_MS * 2)
461-
expect(resolveQuotaContinuationDelayMs(Number.MAX_SAFE_INTEGER)).toBe(6 * 60 * 60 * 1000)
488+
const first = resolveQuotaContinuationDelayMs(1)
489+
const second = resolveQuotaContinuationDelayMs(2)
490+
const capped = resolveQuotaContinuationDelayMs(Number.MAX_SAFE_INTEGER)
491+
492+
expect(first).toBeGreaterThanOrEqual(EMBEDDING_QUOTA_CIRCUIT_TTL_MS * 0.8)
493+
expect(first).toBeLessThanOrEqual(EMBEDDING_QUOTA_CIRCUIT_TTL_MS * 1.2)
494+
expect(second).toBeGreaterThanOrEqual(EMBEDDING_QUOTA_CIRCUIT_TTL_MS * 2 * 0.8)
495+
expect(second).toBeLessThanOrEqual(EMBEDDING_QUOTA_CIRCUIT_TTL_MS * 2 * 1.2)
496+
expect(capped).toBeGreaterThanOrEqual(6 * 60 * 60 * 1000 * 0.8)
497+
expect(capped).toBeLessThanOrEqual(6 * 60 * 60 * 1000)
462498
})
463499
})

‎apps/sim/background/knowledge-processing.ts‎

Lines changed: 13 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,8 @@ import {
99
type DocumentProcessingPayload,
1010
} from '@/lib/knowledge/documents/processing-payload'
1111
import {
12+
canScheduleDocumentProcessingQuotaContinuation,
13+
MAX_QUOTA_CONTINUATION_ATTEMPTS,
1214
resolveQuotaContinuationDelayMs,
1315
scheduleDocumentProcessingQuotaContinuation,
1416
} from '@/lib/knowledge/documents/processing-quota-continuation'
@@ -37,6 +39,7 @@ export async function runDocumentProcessing(
3739
actorUserId: payload.actorUserId,
3840
workspaceId: null,
3941
}
42+
const canScheduleQuotaContinuation = canScheduleDocumentProcessingQuotaContinuation(payload)
4043

4144
logger.info(`[${requestId}] Starting Trigger.dev processing for document: ${docData.filename}`)
4245

@@ -59,7 +62,11 @@ export async function runDocumentProcessing(
5962
...(payload.processingQueuedAt
6063
? { processingQueuedAt: new Date(payload.processingQueuedAt) }
6164
: {}),
62-
scheduleQuotaContinuation: () => scheduleDocumentProcessingQuotaContinuation(payload),
65+
...(canScheduleQuotaContinuation
66+
? {
67+
scheduleQuotaContinuation: () => scheduleDocumentProcessingQuotaContinuation(payload),
68+
}
69+
: { quotaContinuationExhausted: true }),
6370
}
6471
)
6572

@@ -73,13 +80,16 @@ export async function runDocumentProcessing(
7380
}
7481
} catch (error) {
7582
if (isEmbeddingQuotaExhaustion(error)) {
76-
logger.warn(`[${requestId}] Embedding quota is exhausted; scheduling a continuation`, {
83+
const outcome = canScheduleQuotaContinuation ? 'quota_deferred' : 'quota_exhausted'
84+
logger.warn(`[${requestId}] Embedding quota is exhausted`, {
7785
filename: docData.filename,
7886
quotaRetryCount: payload.quotaRetryCount ?? 0,
87+
continuationLimit: MAX_QUOTA_CONTINUATION_ATTEMPTS,
88+
outcome,
7989
})
8090
return {
8191
success: false,
82-
outcome: 'quota_deferred' as const,
92+
outcome,
8393
documentId,
8494
filename: docData.filename,
8595
error: EMBEDDING_QUOTA_EXHAUSTED_MESSAGE,

‎apps/sim/connectors/fireflies/fireflies.test.ts‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -316,13 +316,13 @@ describe('fireflies listDocuments', () => {
316316
expect(global.fetch).toHaveBeenCalledTimes(2)
317317
})
318318

319-
it('surfaces the errors[] message on a non-2xx response', async () => {
319+
it('surfaces a validated errors[] code without the provider message', async () => {
320320
mockGraphQL([
321321
{ status: 403, body: { errors: [{ message: 'Upgrade required', code: 'paid_required' }] } },
322322
])
323323

324324
await expect(firefliesConnector.listDocuments('key', {}, undefined, {})).rejects.toThrow(
325-
/Upgrade required/
325+
/paid_required/
326326
)
327327
expect(global.fetch).toHaveBeenCalledTimes(1)
328328
})

‎apps/sim/connectors/fireflies/fireflies.ts‎

Lines changed: 8 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,5 @@
11
import { createLogger } from '@sim/logger'
22
import { getErrorMessage, toError } from '@sim/utils/errors'
3-
import { truncate } from '@sim/utils/string'
4-
import { redactExactSensitiveValues } from '@/lib/core/security/redaction'
53
import { isPayloadSizeLimitError, readResponseTextWithLimit } from '@/lib/core/utils/stream-limits'
64
import {
75
isRetryableError,
@@ -25,7 +23,6 @@ const FIREFLIES_GRAPHQL_URL = 'https://api.fireflies.ai/graphql'
2523
const TRANSCRIPTS_PER_PAGE = 50
2624
const FIREFLIES_MAX_RESPONSE_BYTES = 16 * 1024 * 1024
2725
const FIREFLIES_MAX_EXTRACTED_CONTENT_BYTES = 8 * 1024 * 1024
28-
const FIREFLIES_MAX_ERROR_MESSAGE_CHARS = 2000
2926
const FIREFLIES_DEFAULT_MAX_RETRY_DELAY_MS = 120_000
3027
const FIREFLIES_DEFAULT_MAX_RETRIES = 2
3128
const FIREFLIES_EXTRACTED_CONTENT_SKIP_REASON =
@@ -198,29 +195,25 @@ async function firefliesGraphQL(
198195

199196
const firstError = data?.errors?.[0]
200197
if (firstError) {
201-
const code = firstError.code ? ` (${firstError.code})` : ''
202-
const safeMessage = truncate(
203-
redactExactSensitiveValues(firstError.message || 'Unknown GraphQL error', [accessToken]),
204-
FIREFLIES_MAX_ERROR_MESSAGE_CHARS
205-
)
198+
const safeCode =
199+
typeof firstError.code === 'string' && /^[A-Za-z0-9_.-]{1,64}$/.test(firstError.code)
200+
? firstError.code
201+
: undefined
202+
const code = safeCode ? ` (${safeCode})` : ''
206203
const retryAt = firstError.extensions?.metadata?.retryAfter
207204
const graphQLRetryAfterMs =
208205
typeof retryAt === 'number' && retryAt > Date.now() ? retryAt - Date.now() : undefined
209206
throw new FirefliesApiError(
210-
`Fireflies API error${code}: ${safeMessage}`,
211-
firstError.code,
207+
`Fireflies API error${code}`,
208+
safeCode,
212209
firstError.extensions?.status ?? response.status,
213210
graphQLRetryAfterMs ?? headerRetryAfterMs
214211
)
215212
}
216213

217214
if (!response.ok) {
218-
const safeDiagnostic = truncate(
219-
redactExactSensitiveValues(rawBody, [accessToken]),
220-
FIREFLIES_MAX_ERROR_MESSAGE_CHARS
221-
)
222215
throw new FirefliesApiError(
223-
`Fireflies API HTTP error: ${response.status}${safeDiagnostic ? `: ${safeDiagnostic}` : ''}`,
216+
`Fireflies API HTTP error: ${response.status} (response body omitted)`,
224217
undefined,
225218
response.status,
226219
headerRetryAfterMs

‎apps/sim/connectors/google-docs/google-docs.test.ts‎

Lines changed: 28 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -47,6 +47,27 @@ describe('googleDocsConnector', () => {
4747
})
4848

4949
describe('getDocument', () => {
50+
it('authoritatively skips a listed document that changed type', async () => {
51+
vi.stubGlobal(
52+
'fetch',
53+
vi.fn(
54+
async () =>
55+
new Response(
56+
JSON.stringify({ ...DRIVE_FILE, mimeType: 'application/pdf', trashed: false }),
57+
{ status: 200 }
58+
)
59+
)
60+
)
61+
62+
await expect(
63+
googleDocsConnector.getDocument(ACCESS_TOKEN, {}, DOCUMENT_ID)
64+
).resolves.toMatchObject({
65+
content: '',
66+
skippedReason: 'File is no longer a Google Doc',
67+
skippedExistingDisposition: 'replace',
68+
})
69+
})
70+
5071
it('requests the tab response view and extracts nested tab content', async () => {
5172
const fetchMock = stubFetchDocument(
5273
new Response(
@@ -131,7 +152,7 @@ describe('googleDocsConnector', () => {
131152
expect(docsUrl.searchParams.get('fields')).toBe('tabs')
132153
})
133154

134-
it('includes the structured Google API error message when hydration fails', async () => {
155+
it('omits the structured Google API error message when hydration fails', async () => {
135156
stubFetchDocument(
136157
new Response(
137158
JSON.stringify({
@@ -146,11 +167,11 @@ describe('googleDocsConnector', () => {
146167
)
147168

148169
await expect(googleDocsConnector.getDocument(ACCESS_TOKEN, {}, DOCUMENT_ID)).rejects.toThrow(
149-
`Failed to fetch Google Doc content ${DOCUMENT_ID}: 400 — Invalid field selection tabs`
170+
`Failed to fetch Google Doc content ${DOCUMENT_ID}: 400`
150171
)
151172
})
152173

153-
it('preserves a structured Google API error message from an envelope larger than 2KB', async () => {
174+
it('omits a structured Google API error message from an envelope larger than 2KB', async () => {
154175
stubFetchDocument(
155176
new Response(
156177
JSON.stringify({
@@ -166,11 +187,11 @@ describe('googleDocsConnector', () => {
166187
)
167188

168189
await expect(googleDocsConnector.getDocument(ACCESS_TOKEN, {}, DOCUMENT_ID)).rejects.toThrow(
169-
`Failed to fetch Google Doc content ${DOCUMENT_ID}: 400 — Invalid field selection tabs`
190+
`Failed to fetch Google Doc content ${DOCUMENT_ID}: 400`
170191
)
171192
})
172193

173-
it('redacts credentials from bounded Google API diagnostics', async () => {
194+
it('omits credentials from bounded Google API diagnostics', async () => {
174195
stubFetchDocument(
175196
new Response(
176197
JSON.stringify({
@@ -185,7 +206,7 @@ describe('googleDocsConnector', () => {
185206
)
186207

187208
await expect(googleDocsConnector.getDocument(ACCESS_TOKEN, {}, DOCUMENT_ID)).rejects.toThrow(
188-
`Failed to fetch Google Doc content ${DOCUMENT_ID}: 403 — Authorization: Bearer [REDACTED]`
209+
`Failed to fetch Google Doc content ${DOCUMENT_ID}: 403`
189210
)
190211
})
191212

@@ -200,7 +221,7 @@ describe('googleDocsConnector', () => {
200221
(caught) => caught as Error
201222
)
202223

203-
expect(error?.message).toContain('[REDACTED]')
224+
expect(error?.message).toBe(`Failed to fetch Google Doc content ${DOCUMENT_ID}: 400`)
204225
expect(error?.message).not.toContain(secret)
205226
})
206227

‎apps/sim/connectors/google-docs/google-docs.ts‎

Lines changed: 9 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -1,12 +1,8 @@
11
import { createLogger } from '@sim/logger'
22
import { toError } from '@sim/utils/errors'
3-
import { isRecordLike } from '@sim/utils/object'
4-
import { truncate } from '@sim/utils/string'
5-
import { redactSensitiveValues } from '@/lib/core/security/redaction'
63
import {
74
fetchWithRetry,
8-
readBoundedHttpErrorPayload,
9-
sanitizeHttpErrorDiagnostic,
5+
readBoundedHttpErrorBody,
106
VALIDATE_RETRY_OPTIONS,
117
} from '@/lib/knowledge/documents/utils'
128
import { googleDocsConnectorMeta } from '@/connectors/google-docs/meta'
@@ -149,27 +145,8 @@ interface DocsDocument {
149145

150146
/** Describes a Google API failure without leaking an unbounded response body. */
151147
async function describeGoogleApiFailure(response: Response): Promise<string> {
152-
const payload = await readBoundedHttpErrorPayload(response)
153-
if (!payload.ok) return String(response.status)
154-
155-
const rawBody = payload.body
156-
const normalizedBody = sanitizeHttpErrorDiagnostic(rawBody).replace(/\s+/g, ' ').trim()
157-
if (!normalizedBody) return String(response.status)
158-
159-
try {
160-
const parsed: unknown = JSON.parse(rawBody)
161-
if (isRecordLike(parsed) && isRecordLike(parsed.error)) {
162-
const message = parsed.error.message
163-
if (typeof message === 'string' && message.trim()) {
164-
const safeMessage = redactSensitiveValues(message).replace(/\s+/g, ' ').trim()
165-
return `${response.status} — ${truncate(safeMessage, 500)}`
166-
}
167-
}
168-
} catch {
169-
return `${response.status} — ${truncate(normalizedBody, 500)}`
170-
}
171-
172-
return `${response.status} — ${truncate(normalizedBody, 500)}`
148+
await readBoundedHttpErrorBody(response)
149+
return String(response.status)
173150
}
174151

175152
/**
@@ -572,7 +549,12 @@ export const googleDocsConnector: ConnectorConfig = {
572549
const file = (await response.json()) as DriveFile & { trashed?: boolean }
573550

574551
if (file.trashed) return null
575-
if (file.mimeType !== GOOGLE_DOC_MIME_TYPE) return null
552+
if (file.mimeType !== GOOGLE_DOC_MIME_TYPE) {
553+
return {
554+
...markSkipped(fileToStub(file), 'File is no longer a Google Doc'),
555+
skippedExistingDisposition: 'replace',
556+
}
557+
}
576558

577559
try {
578560
const { content, hasUnresolvedContent } = await fetchDocContent(accessToken, file.id)

‎apps/sim/connectors/google-drive/google-drive-errors.ts‎

Lines changed: 3 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,7 @@
1-
import { truncate } from '@sim/utils/string'
2-
import { redactSensitiveValues } from '@/lib/core/security/redaction'
31
import { readBodyWithLimit } from '@/connectors/utils'
42

53
const GOOGLE_ERROR_BODY_MAX_BYTES = 64 * 1024
6-
const GOOGLE_ERROR_MESSAGE_MAX_LENGTH = 500
74
const GOOGLE_ERROR_REASON_MAX_COUNT = 16
8-
const GOOGLE_ERROR_REASON_MAX_LENGTH = 100
95

106
const EXPORT_TOO_LARGE_REASONS = new Set(['exportSizeLimitExceeded'])
117
const PERMISSION_REASONS = new Set([
@@ -78,24 +74,9 @@ function parseErrorBody(value: unknown): ParsedGoogleErrorBody | undefined {
7874
}
7975
}
8076

81-
function normalizeMessage(message: string | undefined): string | undefined {
82-
if (!message) return undefined
83-
const singleLine = message
84-
.replace(/[\u0000-\u001f\u007f]+/g, ' ')
85-
.replace(/\s+/g, ' ')
86-
.trim()
87-
return singleLine
88-
? truncate(redactSensitiveValues(singleLine), GOOGLE_ERROR_MESSAGE_MAX_LENGTH, '')
89-
: undefined
90-
}
91-
9277
function normalizeReason(reason: string): string | undefined {
93-
const singleLine = reason
94-
.replace(/[\u0000-\u001f\u007f]+/g, ' ')
95-
.replace(/\s+/g, ' ')
96-
.trim()
97-
if (!singleLine) return undefined
98-
return truncate(redactSensitiveValues(singleLine), GOOGLE_ERROR_REASON_MAX_LENGTH, '')
78+
const normalized = reason.trim()
79+
return /^[A-Za-z][A-Za-z0-9_.-]{0,99}$/.test(normalized) ? normalized : undefined
9980
}
10081

10182
function classifyGoogleDriveError(
@@ -158,14 +139,10 @@ export async function readGoogleDriveApiError(response: Response): Promise<Googl
158139
0,
159140
GOOGLE_ERROR_REASON_MAX_COUNT
160141
)
161-
const providerMessage = normalizeMessage(
162-
parsedBody?.error?.message ?? entries.find((entry) => entry.message)?.message
163-
)
164-
165142
return new GoogleDriveApiError(
166143
response.status,
167144
reasons,
168145
classifyGoogleDriveError(response.status, rawReasons),
169-
providerMessage
146+
undefined
170147
)
171148
}

0 commit comments

Comments
 (0)