diff --git a/apps/sim/lib/file-parsers/pdf-parser-cancellation.test.ts b/apps/sim/lib/file-parsers/pdf-parser-cancellation.test.ts index 28259979fef..908e9bdc3fe 100644 --- a/apps/sim/lib/file-parsers/pdf-parser-cancellation.test.ts +++ b/apps/sim/lib/file-parsers/pdf-parser-cancellation.test.ts @@ -154,4 +154,60 @@ describe('PdfParser cancellation', () => { expect(pdf.getPage).not.toHaveBeenCalled() expect(pdf.destroy).toHaveBeenCalledOnce() }) + + it('lets timers run between CPU-bound pages instead of holding the event loop for the whole document', async () => { + const pageCount = 12 + let timerFired = false + let timerFiredBeforeLastPage = false + const { pdf } = pdfWithPageText(pageCount, (pageNumber) => { + const busyUntil = performance.now() + 5 + while (performance.now() < busyUntil) {} + if (pageNumber === pageCount) timerFiredBeforeLastPage = timerFired + return 'page text' + }) + mockOpenPdfDocument.mockResolvedValueOnce(pdf) + + setTimeout(() => { + timerFired = true + }, 0) + await new PdfParser().parseBuffer(Buffer.from('%PDF-1.4')) + + expect(timerFiredBeforeLastPage).toBe(true) + }) + + it('lets timers run while one page streams a huge content stream', async () => { + const chunkCount = 12 + let timerFired = false + let timerFiredBeforeLastChunk = false + let chunksRead = 0 + const reader = { + cancel: vi.fn().mockResolvedValue(undefined), + read: vi.fn(async () => { + const busyUntil = performance.now() + 5 + while (performance.now() < busyUntil) {} + chunksRead++ + if (chunksRead === chunkCount) timerFiredBeforeLastChunk = timerFired + if (chunksRead > chunkCount) return { done: true } + return { value: { items: [{ str: `chunk ${chunksRead} ` }] }, done: false } + }), + } + const page = { + cleanup: vi.fn(), + streamTextContent: vi.fn(() => ({ getReader: () => reader })), + } + const pdf = { + destroy: vi.fn().mockResolvedValue(undefined), + getPage: vi.fn().mockResolvedValue(page), + numPages: 1, + } + mockOpenPdfDocument.mockResolvedValueOnce(pdf) + + setTimeout(() => { + timerFired = true + }, 0) + const result = await new PdfParser().parseBuffer(Buffer.from('%PDF-1.4')) + + expect(timerFiredBeforeLastChunk).toBe(true) + expect(result.content).toContain(`chunk ${chunkCount}`) + }) }) diff --git a/apps/sim/lib/file-parsers/pdf-parser.ts b/apps/sim/lib/file-parsers/pdf-parser.ts index a599f3d7806..ed249e2ad17 100644 --- a/apps/sim/lib/file-parsers/pdf-parser.ts +++ b/apps/sim/lib/file-parsers/pdf-parser.ts @@ -1,6 +1,6 @@ +import { setImmediate as yieldToEventLoop } from 'node:timers/promises' import { readFile } from 'fs/promises' import { createLogger } from '@sim/logger' -import { sleep } from '@sim/utils/helpers' import type { PDFDocumentProxy, PDFPageProxy } from 'pdfjs-dist/types/src/pdf' import { FileParserError } from '@/lib/file-parsers/errors' import { type PdfPageLines, suppressFurniture } from '@/lib/file-parsers/pdf-furniture' @@ -51,8 +51,15 @@ const PDF_EXTRACTION_TIMEOUT_MS = 60_000 */ const MAX_LINE_DECORATION_BYTES = 5 -/** Pages assembled between event-loop yields, so a long document cannot block the loop. */ -const ASSEMBLY_YIELD_EVERY_PAGES = 32 +/** + * pdf.js runs its "fake worker" on this thread and delivers every message as a + * microtask, so a loop that only awaits pdf.js never yields to timers or I/O: a + * large PDF, or one page with a huge content stream, would hold the event loop + * for the whole extraction, freezing every other request on the process. Page + * reads, text-stream reads, and each whole-document assembly pass yield a + * macrotask once this much time has passed since the last yield. + */ +const EVENT_LOOP_SLICE_MS = 10 /** Pages are joined with a paragraph break. */ const PAGE_SEPARATOR = '\n\n' @@ -86,6 +93,16 @@ interface BoundedExtraction { truncated: boolean } +/** Returns a check that yields a macrotask once {@link EVENT_LOOP_SLICE_MS} has passed since its last yield. */ +function createEventLoopSlicer(): () => Promise { + let lastYieldAt = performance.now() + return async () => { + if (performance.now() - lastYieldAt < EVENT_LOOP_SLICE_MS) return + await yieldToEventLoop() + lastYieldAt = performance.now() + } +} + function waitForAbort( operation: Promise, signal?: AbortSignal, @@ -164,6 +181,7 @@ async function readPageWithinBudget( page: PDFPageProxy, budget: number, deadline: number, + yieldIfDue: () => Promise, signal?: AbortSignal ): Promise { signal?.throwIfAborted() @@ -189,6 +207,7 @@ async function readPageWithinBudget( * zero: text that ends exactly on the budget is complete, not truncated. */ while (!dropped) { + await yieldIfDue() const result = await waitForDeadline( reader.read(), deadline, @@ -269,25 +288,30 @@ async function assemblePages( pages: readonly PdfPageLines[], complete: boolean, signal: AbortSignal | undefined, - maxTextBytes: number + maxTextBytes: number, + yieldIfDue: () => Promise ): Promise { signal?.throwIfAborted() const filteredPages = suppressFurniture(pages) + await yieldIfDue() const allLines = filteredPages.flat() const bodyHeight = dominantLineHeight(allLines) + await yieldIfDue() + const compounds = collectCompounds(allLines) + await yieldIfDue() + const words = collectWords(allLines) + await yieldIfDue() const options = { - compounds: collectCompounds(allLines), - words: collectWords(allLines), + compounds, + words, bodyHeight, headingMarkers: PDF_HEADING_MARKERS_ENABLED && headingMarkersViable(allLines, bodyHeight), } const pageTexts: string[] = [] let outputBytes = 0 - for (const [index, lines] of filteredPages.entries()) { - if (index > 0 && index % ASSEMBLY_YIELD_EVERY_PAGES === 0) { - await sleep(0) - signal?.throwIfAborted() - } + for (const lines of filteredPages) { + await yieldIfDue() + signal?.throwIfAborted() const joined = joinLines(lines, options) const text = complete ? normalizePdfWhitespace(sanitizeTextForUTF8(joined)).trim() : joined if (text.length > 0) { @@ -343,7 +367,9 @@ async function extractTextWithinBudget( ) } + const yieldIfDue = createEventLoopSlicer() for (let pageNumber = 1; pageNumber <= pageLimit; pageNumber++) { + await yieldIfDue() signal?.throwIfAborted() const pagePromise = pdf.getPage(pageNumber) const cleanupLatePage = () => { @@ -367,7 +393,7 @@ async function extractTextWithinBudget( let extraction: PageExtraction const pageCharLimit = complete ? MAX_COMPLETE_PDF_PAGE_CHARS : remainingChars try { - extraction = await readPageWithinBudget(page, pageCharLimit, deadline, signal) + extraction = await readPageWithinBudget(page, pageCharLimit, deadline, yieldIfDue, signal) } finally { page.cleanup() } @@ -408,7 +434,7 @@ async function extractTextWithinBudget( } } - let text = await assemblePages(pages, complete, signal, maxTextBytes) + let text = await assemblePages(pages, complete, signal, maxTextBytes, yieldIfDue) /** Paragraph breaks land after the budget is spent; trimming that overflow is a truncation too. */ if (!complete && text.length > MAX_PDF_TEXT_CHARS) { diff --git a/apps/sim/lib/table/orchestration/import.test.ts b/apps/sim/lib/table/orchestration/import.test.ts index 928c9af7e44..56a17ee9113 100644 --- a/apps/sim/lib/table/orchestration/import.test.ts +++ b/apps/sim/lib/table/orchestration/import.test.ts @@ -161,6 +161,50 @@ describe('performTableCsvImport', () => { expect(mockImportAppendRows).not.toHaveBeenCalled() }) + it('reports the row limit for an over-limit file before contending for the job slot', async () => { + mockMarkTableJobRunning.mockResolvedValue(false) + mockGetMaxRowsPerTable.mockResolvedValue(11) + const threeRows = `${CSV}g@h.i,Gus\n` + + const append = await performTableCsvImport(importParams()) + const replaceWithinLimit = await performTableCsvImport( + importParams({ mode: 'replace', fileStream: csvStream(threeRows) }) + ) + mockGetMaxRowsPerTable.mockResolvedValue(2) + const replaceOverLimit = await performTableCsvImport( + importParams({ mode: 'replace', fileStream: csvStream(threeRows) }) + ) + + expect(append).toMatchObject({ success: false, errorCode: 'validation' }) + expect(append.error).toContain('Append would exceed table row limit') + expect(replaceWithinLimit).toMatchObject({ success: false, errorCode: 'conflict' }) + expect(replaceOverLimit).toMatchObject({ success: false, errorCode: 'validation' }) + expect(replaceOverLimit.error).toContain('Replace would exceed table row limit (2)') + }) + + it('lets timers run while coercing a large file instead of holding the event loop', async () => { + const rowCount = 50_000 + const lines = ['email,name'] + for (let i = 0; i < rowCount; i++) lines.push(`user${i}@example.com,Name ${i}`) + let timerFired = false + let timerFiredBeforeWrite = false + mockGetMaxRowsPerTable.mockResolvedValue(-1) + mockImportAppendRows.mockImplementation(async () => { + timerFiredBeforeWrite = timerFired + return { inserted: [], table: TABLE } + }) + + setTimeout(() => { + timerFired = true + }, 0) + const result = await performTableCsvImport( + importParams({ fileStream: csvStream(`${lines.join('\n')}\n`) }) + ) + + expect(result.success).toBe(true) + expect(timerFiredBeforeWrite).toBe(true) + }) + it('rejects an archived table and a table with a job already running', async () => { const archived = await performTableCsvImport( importParams({ table: { ...TABLE, archivedAt: new Date() } }) diff --git a/apps/sim/lib/table/orchestration/import.ts b/apps/sim/lib/table/orchestration/import.ts index 5b2dc97549b..badac316e35 100644 --- a/apps/sim/lib/table/orchestration/import.ts +++ b/apps/sim/lib/table/orchestration/import.ts @@ -1,4 +1,5 @@ import type { Readable } from 'node:stream' +import { setImmediate as yieldToEventLoop } from 'node:timers/promises' import { createLogger } from '@sim/logger' import { toError } from '@sim/utils/errors' import { generateId } from '@sim/utils/id' @@ -267,6 +268,27 @@ export interface PerformTableCsvImportResult { data?: TableCsvImportData } +/** Rows coerced between event-loop yields; one slice of a wide sheet stays in single-digit ms. */ +const COERCE_SLICE_ROWS = 2_000 + +/** + * {@link coerceRowsForTable} over a whole synchronous-import file, yielding between + * slices. One `rows.map` over a 25 MB file holds the event loop for up to a second, + * stalling every other request the process is serving. + */ +async function coerceRowsInSlices( + ...[rows, ...rest]: Parameters +): Promise> { + const coerced: ReturnType = [] + for (let start = 0; start < rows.length; start += COERCE_SLICE_ROWS) { + if (start > 0) await yieldToEventLoop() + for (const row of coerceRowsForTable(rows.slice(start, start + COERCE_SLICE_ROWS), ...rest)) { + coerced.push(row) + } + } + return coerced +} + /** * Imports a CSV into an EXISTING table, appending or replacing its rows. * @@ -335,8 +357,20 @@ export async function performTableCsvImport( ) } + const maxRows = await getMaxRowsPerTable(workspaceId) + const currentRowCount = mode === 'append' ? table.rowCount : 0 + if (wouldExceedRowLimit(maxRows, currentRowCount, rows.length)) { + const deficit = currentRowCount + rows.length - maxRows + return fail( + mode === 'append' + ? `Append would exceed table row limit (${maxRows}). Currently ${table.rowCount} rows, ${rows.length} new rows, ${deficit} over.` + : `Replace would exceed table row limit (${maxRows}). The file has ${rows.length} rows, ${deficit} over.`, + 'validation' + ) + } + let cellsRejected = 0 - const coerced = coerceRowsForTable( + const coerced = await coerceRowsInSlices( rows, prospectiveSchema, validation.effectiveMap, @@ -371,15 +405,6 @@ export async function performTableCsvImport( try { if (mode === 'append') { - const maxRows = await getMaxRowsPerTable(workspaceId) - if (wouldExceedRowLimit(maxRows, table.rowCount, coerced.length)) { - const deficit = table.rowCount + coerced.length - maxRows - return fail( - `Append would exceed table row limit (${maxRows}). Currently ${table.rowCount} rows, ${coerced.length} new rows, ${deficit} over.`, - 'validation' - ) - } - const { inserted, table: finalTable } = await importAppendRows(table, additions, coerced, { workspaceId, userId,