Skip to content

Commit 7533745

Browse files
committed
improvement(file-parsers): parse spreadsheets on a worker thread
XLSX.read is synchronous and blocks the event loop for the full parse (about 1.1-2s on 16-23 MB workbooks). Spreadsheet buffers going through parseBuffer now run on a bounded worker_threads pool (one worker) built from the same XlsxParser, with an in-process fallback for a missing or broken bundle, worker startup failure, or crash.
1 parent 7992b56 commit 7533745

9 files changed

Lines changed: 665 additions & 7 deletions

File tree

‎apps/sim/lib/file-parsers/index.ts‎

Lines changed: 19 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -26,11 +26,24 @@ import type {
2626
SupportedFileType,
2727
} from '@/lib/file-parsers/types'
2828
import { XlsxParser } from '@/lib/file-parsers/xlsx-parser'
29+
import { parseXlsxBufferOffThread } from '@/lib/file-parsers/xlsx-worker-pool'
2930
import { parseYAML, parseYAMLBuffer } from '@/lib/file-parsers/yaml-parser'
3031
import { assertOoxmlArchiveWithinLimits } from '@/lib/file-parsers/zip-guard'
3132

3233
const logger = createLogger('FileParser')
3334

35+
const xlsxParser = new XlsxParser()
36+
37+
/**
38+
* Spreadsheet buffers parse on a worker thread (see `xlsx-worker-pool.ts`); the
39+
* in-process parser stays the fallback and the `parseFile` path.
40+
*/
41+
const SPREADSHEET_PARSER: FileParser = {
42+
parseFile: (filePath, options) => xlsxParser.parseFile(filePath, options),
43+
parseBuffer: (buffer, options = {}) =>
44+
parseXlsxBufferOffThread(buffer, options, () => xlsxParser.parseBuffer(buffer, options)),
45+
}
46+
3447
/**
3548
* Extension → parser. Several extensions deliberately share one parser because
3649
* they are the same container:
@@ -65,12 +78,12 @@ const PARSERS = new Map<string, FileParser>([
6578
['dotx', new DocxParser()],
6679
['txt', new TxtParser()],
6780
['md', new MdParser()],
68-
['xlsx', new XlsxParser()],
69-
['xls', new XlsxParser()],
70-
['xlsm', new XlsxParser()],
71-
['xlsb', new XlsxParser()],
72-
['xltx', new XlsxParser()],
73-
['ods', new XlsxParser()],
81+
['xlsx', SPREADSHEET_PARSER],
82+
['xls', SPREADSHEET_PARSER],
83+
['xlsm', SPREADSHEET_PARSER],
84+
['xlsb', SPREADSHEET_PARSER],
85+
['xltx', SPREADSHEET_PARSER],
86+
['ods', SPREADSHEET_PARSER],
7487
['pptx', new PptxParser()],
7588
['pptm', new PptxParser()],
7689
['potx', new PptxParser()],
Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,37 @@
1+
/**
2+
* Worker-thread entry for spreadsheet parsing, bundled to
3+
* `lib/file-parsers/dist/xlsx-worker.mjs` by `bun run build:xlsx-worker` and
4+
* spawned by `xlsx-worker-pool.ts`. It runs the same {@link XlsxParser} the
5+
* main thread falls back to, so the two paths cannot drift.
6+
*/
7+
import { parentPort } from 'node:worker_threads'
8+
import { XlsxParser } from '@/lib/file-parsers/xlsx-parser'
9+
import {
10+
serializeXlsxWorkerError,
11+
type XlsxWorkerMessage,
12+
type XlsxWorkerParseRequest,
13+
} from '@/lib/file-parsers/xlsx-worker-protocol'
14+
15+
const port = parentPort
16+
if (!port) throw new Error('xlsx-worker-entry must run inside a worker thread')
17+
18+
const parser = new XlsxParser()
19+
20+
function reply(message: XlsxWorkerMessage) {
21+
port?.postMessage(message)
22+
}
23+
24+
port.on('message', async (request: XlsxWorkerParseRequest) => {
25+
try {
26+
const buffer = Buffer.from(
27+
request.data.buffer,
28+
request.data.byteOffset,
29+
request.data.byteLength
30+
)
31+
reply({ type: 'result', result: await parser.parseBuffer(buffer, request.options) })
32+
} catch (error) {
33+
reply({ type: 'error', error: serializeXlsxWorkerError(error) })
34+
}
35+
})
36+
37+
reply({ type: 'ready' })
Lines changed: 217 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,217 @@
1+
/**
2+
* The parse worker is a cross-process boundary type-check cannot see: results
3+
* and errors are structured-cloned, and the bundle is built by a separate
4+
* script. Failure modes this file guards, each one test:
5+
*
6+
* 1. Worker output diverges from in-process parsing (preview, complete).
7+
* 2. A typed parser error loses its class, `code`, or message crossing back.
8+
* 3. A missing bundle breaks parsing instead of falling back in-process.
9+
* 4. An unloadable bundle (throws at load) breaks parsing instead of falling back.
10+
* 5. A worker crash mid-parse loses the job instead of re-running it in-process.
11+
* 6. An abort during a worker parse hangs or leaves the worker occupying its slot.
12+
* 7. An abort while queued still dispatches the job.
13+
* 8. Dispatch detaches the caller's buffer (transfer of caller-owned memory).
14+
*/
15+
import { execFileSync } from 'node:child_process'
16+
import { mkdtempSync, writeFileSync } from 'node:fs'
17+
import { tmpdir } from 'node:os'
18+
import path from 'node:path'
19+
import { beforeAll, describe, expect, it } from 'vitest'
20+
import * as XLSX from 'xlsx'
21+
import type { FileParseOptions, FileParseResult } from '@/lib/file-parsers/types'
22+
import { XlsxParser } from '@/lib/file-parsers/xlsx-parser'
23+
import { resolveXlsxWorkerPath, XlsxWorkerPool } from '@/lib/file-parsers/xlsx-worker-pool'
24+
25+
const APP_ROOT = path.resolve(__dirname, '..', '..')
26+
const parser = new XlsxParser()
27+
let builtWorkerPath: string
28+
let fakeWorkerDir: string
29+
30+
function workbook(): Buffer {
31+
const book = XLSX.utils.book_new()
32+
XLSX.utils.book_append_sheet(
33+
book,
34+
XLSX.utils.aoa_to_sheet([
35+
['Name', 'Amount', 'When'],
36+
['alpha', 1250.5, new Date(Date.UTC(2024, 0, 2))],
37+
['beta\twith tab', 0.2, 'text'],
38+
]),
39+
'First'
40+
)
41+
XLSX.utils.book_append_sheet(book, XLSX.utils.aoa_to_sheet([['only'], ['cells']]), 'Second')
42+
return XLSX.write(book, { type: 'buffer', bookType: 'xlsx', cellDates: true }) as Buffer
43+
}
44+
45+
/** A ZIP declaring more central-directory records than the archive guard allows. */
46+
function oversizedCentralDirectory(): Buffer {
47+
const entryCount = 10_001
48+
const centralDirectory = Buffer.alloc(46 * entryCount)
49+
for (let index = 0; index < entryCount; index++) {
50+
centralDirectory.writeUInt32LE(0x02014b50, index * 46)
51+
}
52+
const eocd = Buffer.alloc(22)
53+
eocd.writeUInt32LE(0x06054b50, 0)
54+
eocd.writeUInt16LE(entryCount, 8)
55+
eocd.writeUInt16LE(entryCount, 10)
56+
eocd.writeUInt32LE(centralDirectory.length, 12)
57+
return Buffer.concat([centralDirectory, eocd])
58+
}
59+
60+
function writeFakeWorker(name: string, source: string): string {
61+
const file = path.join(fakeWorkerDir, name)
62+
writeFileSync(file, source)
63+
return file
64+
}
65+
66+
type Outcome =
67+
| { ok: true; result: FileParseResult }
68+
| { ok: false; error: { ctor: string; name: string; code: unknown; message: string } }
69+
70+
async function settle(promise: Promise<FileParseResult>): Promise<Outcome> {
71+
try {
72+
return { ok: true, result: await promise }
73+
} catch (error) {
74+
const e = error as Error & { code?: unknown }
75+
return {
76+
ok: false,
77+
error: { ctor: e.constructor.name, name: e.name, code: e.code, message: e.message },
78+
}
79+
}
80+
}
81+
82+
function inProcess(buffer: Buffer, options: FileParseOptions = {}) {
83+
return () => parser.parseBuffer(buffer, options)
84+
}
85+
86+
beforeAll(() => {
87+
execFileSync('bun', ['run', 'build:xlsx-worker'], { cwd: APP_ROOT, stdio: 'pipe' })
88+
const resolved = resolveXlsxWorkerPath()
89+
if (!resolved) throw new Error('build:xlsx-worker did not produce a resolvable bundle')
90+
builtWorkerPath = resolved
91+
fakeWorkerDir = mkdtempSync(path.join(tmpdir(), 'xlsx-worker-pool-'))
92+
}, 60_000)
93+
94+
describe('XlsxWorkerPool against the built bundle', () => {
95+
const cases: Array<{ name: string; buffer: () => Buffer; options: FileParseOptions }> = [
96+
{ name: 'preview workbook', buffer: workbook, options: {} },
97+
{ name: 'complete workbook', buffer: workbook, options: { contentMode: 'complete' } },
98+
{
99+
name: 'complete text budget exceeded',
100+
buffer: workbook,
101+
options: { contentMode: 'complete', maxTextBytes: 8 },
102+
},
103+
{ name: 'archive guard rejection', buffer: oversizedCentralDirectory, options: {} },
104+
{
105+
name: 'corrupt ZIP payload',
106+
buffer: () => Buffer.concat([Buffer.from('PK\u0003\u0004'), Buffer.alloc(256, 7)]),
107+
options: {},
108+
},
109+
]
110+
111+
it.each(cases)('matches in-process parsing for $name', async ({ buffer, options }) => {
112+
const expected = await settle(parser.parseBuffer(buffer(), options))
113+
const pool = new XlsxWorkerPool({ workerPath: builtWorkerPath, size: 1 })
114+
const input = buffer()
115+
116+
const actual = await settle(
117+
pool.parse(input, options, () => Promise.reject(new Error('fell back')))
118+
)
119+
120+
expect(actual).toEqual(expected)
121+
})
122+
123+
it('leaves the caller buffer intact after dispatch', async () => {
124+
const input = workbook()
125+
const snapshot = Buffer.from(input)
126+
const pool = new XlsxWorkerPool({ workerPath: builtWorkerPath, size: 1 })
127+
128+
await pool.parse(input, {}, () => Promise.reject(new Error('fell back')))
129+
130+
expect(input.byteLength).toBe(snapshot.byteLength)
131+
expect(input.equals(snapshot)).toBe(true)
132+
})
133+
})
134+
135+
describe('XlsxWorkerPool fallback', () => {
136+
it('parses in-process when no bundle is available', async () => {
137+
const input = workbook()
138+
const expected = await parser.parseBuffer(input)
139+
const pool = new XlsxWorkerPool({ workerPath: null, size: 1 })
140+
141+
await expect(pool.parse(input, {}, inProcess(input))).resolves.toEqual(expected)
142+
})
143+
144+
it('parses in-process when the bundle throws while loading', async () => {
145+
const input = workbook()
146+
const expected = await parser.parseBuffer(input)
147+
const workerPath = writeFakeWorker('throws.mjs', "throw new Error('broken bundle')\n")
148+
const pool = new XlsxWorkerPool({ workerPath, size: 1 })
149+
150+
await expect(pool.parse(input, {}, inProcess(input))).resolves.toEqual(expected)
151+
await expect(pool.parse(input, {}, inProcess(input))).resolves.toEqual(expected)
152+
})
153+
154+
it('re-runs a job in-process when the worker dies mid-parse', async () => {
155+
const input = workbook()
156+
const expected = await parser.parseBuffer(input)
157+
const workerPath = writeFakeWorker(
158+
'crashes.mjs',
159+
[
160+
"import { parentPort } from 'node:worker_threads'",
161+
"parentPort.on('message', () => process.exit(3))",
162+
"parentPort.postMessage({ type: 'ready' })",
163+
].join('\n')
164+
)
165+
const pool = new XlsxWorkerPool({ workerPath, size: 1 })
166+
167+
await expect(pool.parse(input, {}, inProcess(input))).resolves.toEqual(expected)
168+
})
169+
})
170+
171+
describe('XlsxWorkerPool cancellation', () => {
172+
/** Spins forever on `maxTextBytes: 1`, otherwise answers immediately. */
173+
const spinningWorkerSource = [
174+
"import { parentPort } from 'node:worker_threads'",
175+
"parentPort.on('message', (request) => {",
176+
' if (request.options.maxTextBytes === 1) for (;;) {}',
177+
" parentPort.postMessage({ type: 'result', result: { content: 'answered' } })",
178+
'})',
179+
"parentPort.postMessage({ type: 'ready' })",
180+
].join('\n')
181+
182+
it('rejects with the abort reason and frees the slot held by a spinning worker', async () => {
183+
const workerPath = writeFakeWorker('spins.mjs', spinningWorkerSource)
184+
const pool = new XlsxWorkerPool({ workerPath, size: 1 })
185+
const controller = new AbortController()
186+
const reason = new Error('deadline')
187+
const fallback = () => Promise.reject(new Error('fell back'))
188+
189+
const spinning = pool.parse(
190+
workbook(),
191+
{ signal: controller.signal, maxTextBytes: 1 },
192+
fallback
193+
)
194+
const queued = pool.parse(workbook(), {}, fallback)
195+
setTimeout(() => controller.abort(reason), 50)
196+
197+
await expect(spinning).rejects.toBe(reason)
198+
await expect(queued).resolves.toEqual({ content: 'answered' })
199+
})
200+
201+
it('rejects a queued job on abort while the worker is busy', async () => {
202+
const workerPath = writeFakeWorker('spins-queue.mjs', spinningWorkerSource)
203+
const pool = new XlsxWorkerPool({ workerPath, size: 1 })
204+
const blocker = new AbortController()
205+
const queuedController = new AbortController()
206+
const reason = new Error('caller gave up')
207+
const fallback = () => Promise.reject(new Error('fell back'))
208+
209+
const spinning = pool.parse(workbook(), { signal: blocker.signal, maxTextBytes: 1 }, fallback)
210+
const queued = pool.parse(workbook(), { signal: queuedController.signal }, fallback)
211+
queuedController.abort(reason)
212+
213+
await expect(queued).rejects.toBe(reason)
214+
blocker.abort(new Error('cleanup'))
215+
await expect(spinning).rejects.toThrow('cleanup')
216+
})
217+
})

0 commit comments

Comments
 (0)