fix(parsers): yield the event loop during PDF extraction and CSV row coercion

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}`)
+  })
 })

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<void> {
+  let lastYieldAt = performance.now()
+  return async () => {
+    if (performance.now() - lastYieldAt < EVENT_LOOP_SLICE_MS) return
+    await yieldToEventLoop()
+    lastYieldAt = performance.now()
+  }
+}
+
 function waitForAbort<T>(
   operation: Promise<T>,
   signal?: AbortSignal,
@@ -164,6 +181,7 @@ async function readPageWithinBudget(
   page: PDFPageProxy,
   budget: number,
   deadline: number,
+  yieldIfDue: () => Promise<void>,
   signal?: AbortSignal
 ): Promise<PageExtraction> {
   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<void>
 ): Promise<string> {
   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) {

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() } })

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<typeof coerceRowsForTable>
+): Promise<ReturnType<typeof coerceRowsForTable>> {
+  const coerced: ReturnType<typeof coerceRowsForTable> = []
+  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,