From fabec0aec3924d23bd6722280b0ec5e04f413b30 Mon Sep 17 00:00:00 2001 From: Abimael Martell <1450169+abimaelmartell@users.noreply.github.com> Date: Mon, 10 Aug 2026 12:29:23 -0700 Subject: [PATCH] feat(napi): add async variants that keep the Node event loop free (#337) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(napi): add processPdfAsync, classifyPdfAsync, extractPagesMarkdownAsync The Node bindings are synchronous, so every call parses on the event loop thread — up to hundreds of milliseconds of dead loop per document in a server. Add additive AsyncTask-based variants that run the same shared implementations on the libuv thread pool and return promises. The existing synchronous exports keep their names, signatures, and behaviour; each sync/async pair shares one implementation. Panics in compute() are caught and surfaced as rejections, matching the sync error contract. Closes #336 Co-authored-by: Abimael Martell * fix(napi): read async task buffers in place instead of copying Review feedback on #337: buffer.to_vec() copied the whole PDF on the event loop before the task was queued, so large inputs still stalled the loop and doubled peak memory. The tasks now hold the napi Buffer itself — its ref pins the JS allocation for the task's lifetime and the backing store is stable, so compute() reads it directly from the worker thread. Callers must not mutate the buffer until the promise settles (same contract as Node's async fs APIs); documented on each export and in the README. The suggested removal of ts_return_type was checked and rejected: without it napi-rs generates Promise for AsyncTask returns. A comment now records that finding. Co-authored-by: Abimael Martell * fix(napi): copy async task input on the JS thread for soundness Review feedback on #337: holding the napi Buffer and reading it from the libuv worker was unsound. Buffer derefs straight to the JS-side allocation, so a caller mutating it before the promise settled would race the worker's reads — undefined behavior, not a recoverable error, and the documented don't-mutate contract was unenforceable. Deferring the copy to compute() would not help: any off-thread read races the same way. The JS thread is the only race-free place to take the copy, because JS is single-threaded and nothing can mutate the buffer during the synchronous part of the call. Revert to an owned Vec copied at call time. The cost is one memcpy, negligible next to the parse the async variants exist to unblock. Docs now state the buffer may be reused or mutated immediately, and a test locks in the copy semantics by mutating the input while a parse is in flight. Co-authored-by: Abimael Martell --------- Co-authored-by: Cursor Agent Co-authored-by: Abimael Martell --- napi/README.md | 16 ++++ napi/src/lib.rs | 223 +++++++++++++++++++++++++++++++++++++++--------- napi/test.mjs | 68 +++++++++++++++ 3 files changed, 266 insertions(+), 41 deletions(-) diff --git a/napi/README.md b/napi/README.md index ba6d3f4..414d869 100644 --- a/napi/README.md +++ b/napi/README.md @@ -83,6 +83,22 @@ for (const region of result[0].regions) { } ``` +### Async variants + +`processPdf`, `classifyPdf`, and `extractPagesMarkdown` are synchronous and parse on the calling thread — in Node, that's the event loop. For a one-off call in a script that's fine, but in a server a large document can hold the loop for tens to hundreds of milliseconds. + +`processPdfAsync`, `classifyPdfAsync`, and `extractPagesMarkdownAsync` take the same arguments and produce the same results, but run the parse on the libuv thread pool and return a promise, keeping the event loop free. The input buffer is copied before the call returns, so it's safe to reuse or mutate immediately: + +```typescript +import { classifyPdfAsync, extractPagesMarkdownAsync } from '@firecrawl/pdf-inspector' + +const classification = await classifyPdfAsync(pdf) +if (classification.pdfType === 'TextBased') { + const { pages } = await extractPagesMarkdownAsync(pdf) + // ... +} +``` + ## Types ```typescript diff --git a/napi/src/lib.rs b/napi/src/lib.rs index a2280df..c4d968d 100644 --- a/napi/src/lib.rs +++ b/napi/src/lib.rs @@ -153,9 +153,7 @@ fn to_napi_result(r: pdf_inspector::PdfProcessResult) -> PdfResult { } } -fn to_napi_page_ocr_reasons( - reasons: Vec, -) -> Vec { +fn to_napi_page_ocr_reasons(reasons: Vec) -> Vec { reasons .into_iter() .map(|reason| PageOcrReasons { @@ -202,6 +200,31 @@ where } } +// --------------------------------------------------------------------------- +// Shared implementations (single body behind sync and async entry points) +// --------------------------------------------------------------------------- + +fn process_pdf_impl(bytes: &[u8], pages: Option>) -> Result { + let mut opts = pdf_inspector::PdfOptions::new(); + if let Some(p) = pages { + opts = opts.pages(p); + } + let result = pdf_inspector::process_pdf_mem_with_options(bytes, opts) + .map_err(|e| to_napi_err(e, "process_pdf"))?; + Ok(to_napi_result(result)) +} + +fn classify_pdf_impl(bytes: &[u8]) -> Result { + let result = + pdf_inspector::classify_pdf_mem(bytes).map_err(|e| to_napi_err(e, "classify_pdf"))?; + Ok(PdfClassification { + pdf_type: convert_pdf_type(result.pdf_type), + page_count: result.page_count, + pages_needing_ocr: result.pages_needing_ocr, + confidence: result.confidence as f64, + }) +} + // --------------------------------------------------------------------------- // Public NAPI API // --------------------------------------------------------------------------- @@ -210,15 +233,7 @@ where #[napi] pub fn process_pdf(buffer: Buffer, pages: Option>) -> Result { let bytes: Vec = buffer.to_vec(); - catch_panic("process_pdf", move || { - let mut opts = pdf_inspector::PdfOptions::new(); - if let Some(p) = pages { - opts = opts.pages(p); - } - let result = pdf_inspector::process_pdf_mem_with_options(&bytes, opts) - .map_err(|e| to_napi_err(e, "process_pdf"))?; - Ok(to_napi_result(result)) - }) + catch_panic("process_pdf", move || process_pdf_impl(&bytes, pages)) } /// Fast detection only — no text extraction or markdown. @@ -238,16 +253,7 @@ pub fn detect_pdf(buffer: Buffer) -> Result { #[napi] pub fn classify_pdf(buffer: Buffer) -> Result { let bytes: Vec = buffer.to_vec(); - catch_panic("classify_pdf", move || { - let result = - pdf_inspector::classify_pdf_mem(&bytes).map_err(|e| to_napi_err(e, "classify_pdf"))?; - Ok(PdfClassification { - pdf_type: convert_pdf_type(result.pdf_type), - page_count: result.page_count, - pages_needing_ocr: result.pages_needing_ocr, - confidence: result.confidence as f64, - }) - }) + catch_panic("classify_pdf", move || classify_pdf_impl(&bytes)) } /// Extract plain text from a PDF Buffer. @@ -633,25 +639,32 @@ pub fn extract_pages_markdown( ) -> Result { let bytes: Vec = buffer.to_vec(); catch_panic("extract_pages_markdown", move || { - let result = pdf_inspector::extract_pages_markdown_mem(&bytes, pages.as_deref()) - .map_err(|e| to_napi_err(e, "extract_pages_markdown"))?; - Ok(PagesExtractionResult { - pages: result - .pages - .into_iter() - .map(|r| PageMarkdownResult { - page: r.page, - markdown: r.markdown, - needs_ocr: r.needs_ocr, - ocr_reason: r.ocr_reason, - }) - .collect(), - pages_with_tables: result.pages_with_tables, - pages_with_columns: result.pages_with_columns, - pages_needing_ocr: result.pages_needing_ocr, - ocr_reasons_by_page: to_napi_page_ocr_reasons(result.ocr_reasons_by_page), - is_complex: result.is_complex, - }) + extract_pages_markdown_impl(&bytes, pages.as_deref()) + }) +} + +fn extract_pages_markdown_impl( + bytes: &[u8], + pages: Option<&[u32]>, +) -> Result { + let result = pdf_inspector::extract_pages_markdown_mem(bytes, pages) + .map_err(|e| to_napi_err(e, "extract_pages_markdown"))?; + Ok(PagesExtractionResult { + pages: result + .pages + .into_iter() + .map(|r| PageMarkdownResult { + page: r.page, + markdown: r.markdown, + needs_ocr: r.needs_ocr, + ocr_reason: r.ocr_reason, + }) + .collect(), + pages_with_tables: result.pages_with_tables, + pages_with_columns: result.pages_with_columns, + pages_needing_ocr: result.pages_needing_ocr, + ocr_reasons_by_page: to_napi_page_ocr_reasons(result.ocr_reasons_by_page), + is_complex: result.is_complex, }) } @@ -692,3 +705,131 @@ fn to_page_region_texts(results: Vec) -> Vec` on the calling +// (JS) thread — deliberately. JS execution is single-threaded, so no JS code +// can mutate the buffer while the synchronous part of the call copies it. +// Holding the napi `Buffer` and reading it from the worker instead would be +// zero-copy, but a caller mutating the buffer before the promise settles +// would then race the worker's reads — undefined behavior, not a recoverable +// error (a known napi-rs soundness hazard with cross-thread Buffer access). +// The copy is a one-time memcpy, negligible next to the parse it unblocks. +// --------------------------------------------------------------------------- + +pub struct ProcessPdfTask { + bytes: Vec, + pages: Option>, +} + +impl Task for ProcessPdfTask { + type Output = PdfResult; + type JsValue = PdfResult; + + fn compute(&mut self) -> Result { + let bytes = std::mem::take(&mut self.bytes); + let pages = self.pages.take(); + // AssertUnwindSafe: `bytes`/`pages` are moved into the closure and + // dropped on unwind — no shared state can be observed broken. + catch_panic( + "process_pdf", + panic::AssertUnwindSafe(move || process_pdf_impl(&bytes, pages)), + ) + } + + fn resolve(&mut self, _env: Env, output: Self::Output) -> Result { + Ok(output) + } +} + +/// Async variant of [`processPdf`]: same result, but the parse runs on the +/// libuv thread pool instead of the event loop and the call returns a +/// promise. The buffer is copied before the call returns, so it may be +/// reused or mutated immediately. +// ts_return_type is required: napi-rs emits `Promise` for +// `AsyncTask` returns without it. +#[napi(ts_return_type = "Promise")] +pub fn process_pdf_async(buffer: Buffer, pages: Option>) -> AsyncTask { + AsyncTask::new(ProcessPdfTask { + bytes: buffer.to_vec(), + pages, + }) +} + +pub struct ClassifyPdfTask { + bytes: Vec, +} + +impl Task for ClassifyPdfTask { + type Output = PdfClassification; + type JsValue = PdfClassification; + + fn compute(&mut self) -> Result { + let bytes = std::mem::take(&mut self.bytes); + catch_panic( + "classify_pdf", + panic::AssertUnwindSafe(move || classify_pdf_impl(&bytes)), + ) + } + + fn resolve(&mut self, _env: Env, output: Self::Output) -> Result { + Ok(output) + } +} + +/// Async variant of [`classifyPdf`]: same result, but the classification runs +/// on the libuv thread pool instead of the event loop and the call returns a +/// promise. The buffer is copied before the call returns, so it may be +/// reused or mutated immediately. +#[napi(ts_return_type = "Promise")] +pub fn classify_pdf_async(buffer: Buffer) -> AsyncTask { + AsyncTask::new(ClassifyPdfTask { + bytes: buffer.to_vec(), + }) +} + +pub struct ExtractPagesMarkdownTask { + bytes: Vec, + pages: Option>, +} + +impl Task for ExtractPagesMarkdownTask { + type Output = PagesExtractionResult; + type JsValue = PagesExtractionResult; + + fn compute(&mut self) -> Result { + let bytes = std::mem::take(&mut self.bytes); + let pages = self.pages.take(); + catch_panic( + "extract_pages_markdown", + panic::AssertUnwindSafe(move || extract_pages_markdown_impl(&bytes, pages.as_deref())), + ) + } + + fn resolve(&mut self, _env: Env, output: Self::Output) -> Result { + Ok(output) + } +} + +/// Async variant of [`extractPagesMarkdown`]: same result, but the extraction +/// runs on the libuv thread pool instead of the event loop and the call +/// returns a promise. The buffer is copied before the call returns, so it +/// may be reused or mutated immediately. +#[napi(ts_return_type = "Promise")] +pub fn extract_pages_markdown_async( + buffer: Buffer, + pages: Option>, +) -> AsyncTask { + AsyncTask::new(ExtractPagesMarkdownTask { + bytes: buffer.to_vec(), + pages, + }) +} diff --git a/napi/test.mjs b/napi/test.mjs index dffc0c7..a020274 100644 --- a/napi/test.mjs +++ b/napi/test.mjs @@ -2,13 +2,16 @@ import { readFileSync } from 'fs'; import { strict as assert } from 'assert'; import { processPdf, + processPdfAsync, detectPdf, classifyPdf, + classifyPdfAsync, extractText, extractTextWithPositions, extractTextInRegions, detectVectorGridInRegion, extractPagesMarkdown, + extractPagesMarkdownAsync, } from './index.js'; const fixture = readFileSync('../tests/fixtures/thermo-freon12.pdf'); @@ -124,10 +127,75 @@ assert.equal(picked.pages[0].page, 2); assert.equal(picked.pages[1].page, 0); console.log(' extractPagesMarkdown with pages: OK'); +// --- Async variants --- +console.log('Testing async variants...'); + +// processPdfAsync returns a promise and matches the sync result +const asyncResultPromise = processPdfAsync(fixture); +assert.ok(asyncResultPromise instanceof Promise); +const asyncResult = await asyncResultPromise; +assert.equal(asyncResult.pdfType, result.pdfType); +assert.equal(asyncResult.pageCount, result.pageCount); +assert.equal(asyncResult.markdown, result.markdown); +console.log(' processPdfAsync: OK'); + +// processPdfAsync with pages +const asyncResult2 = await processPdfAsync(fixture, [1]); +assert.equal(asyncResult2.markdown, result2.markdown); +console.log(' processPdfAsync with pages: OK'); + +// classifyPdfAsync matches the sync result +const asyncClassified = await classifyPdfAsync(fixture); +assert.equal(asyncClassified.pdfType, classified.pdfType); +assert.equal(asyncClassified.pageCount, classified.pageCount); +assert.equal(asyncClassified.confidence, classified.confidence); +assert.deepEqual(asyncClassified.pagesNeedingOcr, classified.pagesNeedingOcr); +console.log(' classifyPdfAsync: OK'); + +// extractPagesMarkdownAsync matches the sync result +const asyncAllPages = await extractPagesMarkdownAsync(fixture); +assert.equal(asyncAllPages.pages.length, allPages.pages.length); +assert.deepEqual( + asyncAllPages.pages.map(p => p.markdown), + allPages.pages.map(p => p.markdown), +); +assert.equal(asyncAllPages.isComplex, allPages.isComplex); +console.log(' extractPagesMarkdownAsync: OK'); + +// selected pages preserve caller order +const asyncPicked = await extractPagesMarkdownAsync(fixture, [2, 0]); +assert.equal(asyncPicked.pages.length, 2); +assert.equal(asyncPicked.pages[0].page, 2); +assert.equal(asyncPicked.pages[1].page, 0); +console.log(' extractPagesMarkdownAsync with pages: OK'); + +// input buffer is copied at call time: mutating it immediately after the +// call must not affect the in-flight parse +const scratch = Buffer.from(fixture); +const inFlight = processPdfAsync(scratch); +scratch.fill(0); +const fromMutated = await inFlight; +assert.equal(fromMutated.markdown, result.markdown); +console.log(' processPdfAsync input copied at call time: OK'); + +// concurrent async calls all settle +const [c1, c2, c3] = await Promise.all([ + processPdfAsync(fixture), + classifyPdfAsync(fixture), + extractPagesMarkdownAsync(fixture), +]); +assert.equal(c1.pdfType, 'TextBased'); +assert.equal(c2.pdfType, 'TextBased'); +assert.equal(c3.pages.length, 3); +console.log(' concurrent async calls: OK'); + // --- Error handling --- console.log('Testing error handling...'); assert.throws(() => processPdf(Buffer.from('not a pdf')), /process_pdf/); assert.throws(() => classifyPdf(Buffer.from('')), /classify_pdf/); +await assert.rejects(processPdfAsync(Buffer.from('not a pdf')), /process_pdf/); +await assert.rejects(classifyPdfAsync(Buffer.from('')), /classify_pdf/); +await assert.rejects(extractPagesMarkdownAsync(Buffer.from('')), /extract_pages_markdown/); console.log(' error handling: OK'); console.log('\nAll NAPI tests passed!');