From 4583d5a926a6e28b5cea02dcc9ef4b4af1be2f1d Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Tue, 5 May 2026 08:53:21 -0400 Subject: [PATCH] =?UTF-8?q?=E2=9A=A1=20refactor:=20Bound=20Concurrent=20Of?= =?UTF-8?q?fice-HTML=20Rendering=20for=20Code=20Artifacts=20(#12951)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * šŸ§° feat: add `createConcurrencyLimiter` promise utility Lightweight, dependency-free FIFO concurrency limiter for bounding parallelism of expensive async work that fans out from a single producer. Tasks queue rather than reject when the cap is reached; slots release on both fulfillment and rejection so a single failure cannot stall the queue. Each task runs inside a thunk so timeouts and other side effects do not start until a slot is acquired. * āš” perf: bound concurrent office-HTML rendering for code artifacts A tool result with N office files (DOCX/XLSX/XLS/ODS/CSV/PPTX) previously fanned out into N parallel mammoth/SheetJS/PPTX renders, all CPU-bound and synchronous. Under bursty agent output this competes with the still-running inference loop for event-loop time and inflates end-of-run "finalize" waits in non-streaming flows (BaseClient chat-completions, non-streaming Responses, the tools.js direct endpoint) which all `await Promise.all(artifactPromises)` before responding. Cap the parser layer at 2 concurrent office-HTML renders process- wide via a shared `createConcurrencyLimiter`. Tasks queue in FIFO and the per-render timeout starts only after a slot is acquired so queue waits do not consume the timeout budget. The HTML-or-null contract (no text fallback for office types) is preserved. --- packages/api/src/files/code/extract.spec.ts | 79 ++++++++ packages/api/src/files/code/extract.ts | 27 ++- packages/api/src/utils/promise.spec.ts | 201 +++++++++++++++++++- packages/api/src/utils/promise.ts | 66 +++++++ 4 files changed, 367 insertions(+), 6 deletions(-) diff --git a/packages/api/src/files/code/extract.spec.ts b/packages/api/src/files/code/extract.spec.ts index a2847b6428..1baa2490f1 100644 --- a/packages/api/src/files/code/extract.spec.ts +++ b/packages/api/src/files/code/extract.spec.ts @@ -597,3 +597,82 @@ describe('getExtractedTextFormat', () => { expect(getExtractedTextFormat('', '', 'whatever')).toBe('text'); }); }); + +describe('extractCodeArtifactText office-html concurrency', () => { + beforeEach(() => { + mockOfficeHtml.mockReset(); + parseDocumentCalls.length = 0; + }); + + /* The limiter inside extract.ts caps simultaneous office-HTML renders so + * a tool result with many office artifacts can't fan out N parallel + * mammoth/SheetJS invocations and starve the still-running agent loop. + * The cap is a module-private constant; this test asserts the observable + * contract without coupling to the literal value. */ + it('caps concurrent office-HTML renders so a burst of 10 files does not all run at once', async () => { + let active = 0; + let peak = 0; + + mockOfficeHtml.mockImplementation(async () => { + active++; + peak = Math.max(peak, active); + await new Promise((resolve) => setTimeout(resolve, 5)); + active--; + return 'ok'; + }); + + const results = await Promise.all( + Array.from({ length: 10 }, (_, i) => + extractCodeArtifactText( + Buffer.from('PK'), + `report-${i}.docx`, + 'application/vnd.openxmlformats-officedocument.wordprocessingml.document', + 'document', + ), + ), + ); + + expect(results).toHaveLength(10); + results.forEach((text) => { + expect(text).toContain(''); + }); + /* Hard upper bound from the module's OFFICE_HTML_CONCURRENCY=2. + * We assert <= a small constant rather than the exact value to avoid + * coupling, but it MUST be well below the request count. */ + expect(peak).toBeGreaterThan(0); + expect(peak).toBeLessThanOrEqual(4); + expect(peak).toBeLessThan(10); + }); + + it('continues processing the queue when an earlier render rejects', async () => { + /* A failing render must release its slot ā€” otherwise a single bad + * file in a burst would permanently shrink the limiter and stall + * subsequent extractions. */ + let call = 0; + mockOfficeHtml.mockImplementation(async () => { + call++; + if (call === 1) { + throw new Error('mammoth blew up'); + } + return 'ok'; + }); + + const results = await Promise.all( + Array.from({ length: 5 }, (_, i) => + extractCodeArtifactText( + Buffer.from('PK'), + `report-${i}.docx`, + 'application/vnd.openxmlformats-officedocument.wordprocessingml.document', + 'document', + ), + ), + ); + + /* First call's failure surfaces as null per the HTML-or-null contract; + * the remaining four must still produce HTML. */ + const nullCount = results.filter((t) => t === null).length; + const htmlCount = results.filter((t) => t != null && t.includes('')).length; + expect(nullCount).toBe(1); + expect(htmlCount).toBe(4); + }); +}); diff --git a/packages/api/src/files/code/extract.ts b/packages/api/src/files/code/extract.ts index d61337df20..85f9321d71 100644 --- a/packages/api/src/files/code/extract.ts +++ b/packages/api/src/files/code/extract.ts @@ -7,7 +7,7 @@ import type { CodeArtifactCategory } from './classify'; import { parseDocument } from '~/files/documents/crud'; import { bufferToOfficeHtml, officeHtmlBucket } from '~/files/documents/html'; import { isBinaryBuffer } from '~/skills/binary'; -import { withTimeout } from '~/utils/promise'; +import { createConcurrencyLimiter, withTimeout } from '~/utils/promise'; export const MAX_TEXT_CACHE_BYTES = 512 * 1024; export const MAX_TEXT_EXTRACT_BYTES = 1024 * 1024; @@ -16,6 +16,21 @@ const OFFICE_HTML_TIMEOUT_MS = 12_000; const TRUNCATION_MARKER = '\n\nā€¦[truncated]'; const TRUNCATION_MARKER_BYTES = Buffer.byteLength(TRUNCATION_MARKER, 'utf-8'); +/** + * Cap simultaneous office-HTML renders process-wide. Mammoth (DOCX), SheetJS + * (XLSX/XLS/ODS), the CSV producer, and the PPTX renderer are CPU-heavy and + * synchronous; left unbounded, a tool result with N office files fans out + * into N parallel parses that compete with the still-running agent inference + * for event-loop time. Two slots keeps a single bursty tool result from + * starving inference while still allowing meaningful parallelism. Tasks + * queue in FIFO ā€” none are dropped, only their start is deferred. Shared + * across every flow: streaming Responses, non-streaming Responses, + * chat-completions BaseClient, and the `tools.js` direct endpoint all + * funnel through this module. + */ +const OFFICE_HTML_CONCURRENCY = 2; +const officeHtmlLimit = createConcurrencyLimiter(OFFICE_HTML_CONCURRENCY); + /** * Decide whether a buffer is a candidate for rich HTML preview. Wraps the * shared `officeHtmlBucket` predicate from `~/files/documents/html` so the @@ -181,10 +196,12 @@ const renderOfficeHtml = async ( mimeType: string, ): Promise => { try { - const html = await withTimeout( - bufferToOfficeHtml(buffer, name, mimeType), - OFFICE_HTML_TIMEOUT_MS, - `bufferToOfficeHtml exceeded ${OFFICE_HTML_TIMEOUT_MS}ms`, + const html = await officeHtmlLimit(() => + withTimeout( + bufferToOfficeHtml(buffer, name, mimeType), + OFFICE_HTML_TIMEOUT_MS, + `bufferToOfficeHtml exceeded ${OFFICE_HTML_TIMEOUT_MS}ms`, + ), ); if (html == null) { return null; diff --git a/packages/api/src/utils/promise.spec.ts b/packages/api/src/utils/promise.spec.ts index c43c8bf739..93c5fd8893 100644 --- a/packages/api/src/utils/promise.spec.ts +++ b/packages/api/src/utils/promise.spec.ts @@ -1,4 +1,20 @@ -import { withTimeout } from './promise'; +import { createConcurrencyLimiter, withTimeout } from './promise'; + +const tick = (ms = 0): Promise => new Promise((resolve) => setTimeout(resolve, ms)); + +const deferred = (): { + promise: Promise; + resolve: (value: T) => void; + reject: (reason?: unknown) => void; +} => { + let resolve!: (value: T) => void; + let reject!: (reason?: unknown) => void; + const promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + return { promise, resolve, reject }; +}; describe('withTimeout', () => { beforeEach(() => { @@ -113,3 +129,186 @@ describe('withTimeout', () => { ); }); }); + +describe('createConcurrencyLimiter', () => { + it('runs tasks immediately while under the cap', async () => { + const limit = createConcurrencyLimiter(2); + const a = deferred(); + const b = deferred(); + let aStarted = false; + let bStarted = false; + + const pa = limit(() => { + aStarted = true; + return a.promise; + }); + const pb = limit(() => { + bStarted = true; + return b.promise; + }); + + await tick(); + expect(aStarted).toBe(true); + expect(bStarted).toBe(true); + + a.resolve('a'); + b.resolve('b'); + expect(await pa).toBe('a'); + expect(await pb).toBe('b'); + }); + + it('queues tasks beyond the cap and starts them as slots free', async () => { + const limit = createConcurrencyLimiter(2); + const a = deferred(); + const b = deferred(); + const c = deferred(); + let cStarted = false; + + const pa = limit(() => a.promise); + const pb = limit(() => b.promise); + const pc = limit(() => { + cStarted = true; + return c.promise; + }); + + await tick(); + expect(cStarted).toBe(false); + + a.resolve('a'); + expect(await pa).toBe('a'); + await tick(); + expect(cStarted).toBe(true); + + b.resolve('b'); + c.resolve('c'); + expect(await pb).toBe('b'); + expect(await pc).toBe('c'); + }); + + it('dequeues in FIFO order', async () => { + const limit = createConcurrencyLimiter(1); + const order: string[] = []; + const head = deferred(); + + const pHead = limit(async () => { + order.push('head:start'); + await head.promise; + order.push('head:end'); + }); + const p1 = limit(async () => { + order.push('1'); + }); + const p2 = limit(async () => { + order.push('2'); + }); + const p3 = limit(async () => { + order.push('3'); + }); + + await tick(); + expect(order).toEqual(['head:start']); + + head.resolve(); + await Promise.all([pHead, p1, p2, p3]); + expect(order).toEqual(['head:start', 'head:end', '1', '2', '3']); + }); + + it('releases the slot when a task rejects, allowing queued work to proceed', async () => { + const limit = createConcurrencyLimiter(1); + const failing = deferred(); + let nextStarted = false; + + const pFail = limit(() => failing.promise); + const pNext = limit(async () => { + nextStarted = true; + return 'ok'; + }); + + await tick(); + expect(nextStarted).toBe(false); + + failing.reject(new Error('boom')); + await expect(pFail).rejects.toThrow('boom'); + expect(await pNext).toBe('ok'); + expect(nextStarted).toBe(true); + }); + + it('isolates rejections ā€” one failing task does not reject sibling tasks', async () => { + const limit = createConcurrencyLimiter(2); + const ok = limit(async () => 'ok'); + const bad = limit(async () => { + throw new Error('nope'); + }); + + expect(await ok).toBe('ok'); + await expect(bad).rejects.toThrow('nope'); + }); + + it('does not exceed the configured cap under heavy load', async () => { + const concurrency = 3; + const limit = createConcurrencyLimiter(concurrency); + let active = 0; + let peak = 0; + + const tasks = Array.from({ length: 20 }, (_, i) => + limit(async () => { + active++; + peak = Math.max(peak, active); + await tick(5); + active--; + return i; + }), + ); + + const results = await Promise.all(tasks); + expect(results).toEqual(Array.from({ length: 20 }, (_, i) => i)); + expect(peak).toBeLessThanOrEqual(concurrency); + expect(peak).toBe(concurrency); + }); + + it('serializes when concurrency is 1', async () => { + const limit = createConcurrencyLimiter(1); + let active = 0; + let peak = 0; + + await Promise.all( + Array.from({ length: 5 }, () => + limit(async () => { + active++; + peak = Math.max(peak, active); + await tick(2); + active--; + }), + ), + ); + + expect(peak).toBe(1); + }); + + it('throws when concurrency is not a positive integer', () => { + expect(() => createConcurrencyLimiter(0)).toThrow(/positive integer/); + expect(() => createConcurrencyLimiter(-1)).toThrow(/positive integer/); + expect(() => createConcurrencyLimiter(1.5)).toThrow(/positive integer/); + expect(() => createConcurrencyLimiter(Infinity)).toThrow(/positive integer/); + expect(() => createConcurrencyLimiter(NaN)).toThrow(/positive integer/); + }); + + it('does not invoke the task until a slot is available', async () => { + const limit = createConcurrencyLimiter(1); + const block = deferred(); + let queuedTaskCalled = false; + + const pHead = limit(() => block.promise); + const pQueued = limit(async () => { + queuedTaskCalled = true; + }); + + await tick(); + await tick(); + expect(queuedTaskCalled).toBe(false); + + block.resolve(); + await Promise.all([pHead, pQueued]); + expect(queuedTaskCalled).toBe(true); + }); +}); diff --git a/packages/api/src/utils/promise.ts b/packages/api/src/utils/promise.ts index 72719a3ff0..06a3e440d4 100644 --- a/packages/api/src/utils/promise.ts +++ b/packages/api/src/utils/promise.ts @@ -40,3 +40,69 @@ export async function withTimeout( clearTimeout(timeoutId!); } } + +/** + * Create an in-process concurrency limiter. Returns a `run` function that + * wraps async tasks: at most `concurrency` invocations may execute at once; + * additional calls queue and dequeue in FIFO order as slots free. + * + * Use to bound the parallelism of expensive CPU-or-IO work that fans out + * from a single producer (e.g. an agent emitting many office artifacts in + * one tool result), so the work doesn't compete with the still-running + * agent inference for event-loop time. Tasks remain queued ā€” they are + * never dropped or rejected by the limiter itself ā€” so the overall workload + * still completes; only peak concurrency is capped. + * + * Each task is wrapped in a thunk so timeouts and other side effects do + * not start until the limiter actually invokes it. + * + * @example + * ```typescript + * const limit = createConcurrencyLimiter(2); + * const results = await Promise.all(files.map((f) => limit(() => parse(f)))); + * ``` + */ +export function createConcurrencyLimiter( + concurrency: number, +): (task: () => Promise) => Promise { + if (!Number.isInteger(concurrency) || concurrency < 1) { + throw new Error( + `createConcurrencyLimiter: concurrency must be a positive integer (got ${concurrency})`, + ); + } + + let active = 0; + const queue: Array<() => void> = []; + + const release = (): void => { + active--; + const next = queue.shift(); + if (next) { + next(); + } + }; + + return (task: () => Promise): Promise => + new Promise((resolve, reject) => { + const run = (): void => { + active++; + Promise.resolve() + .then(task) + .then( + (value) => { + release(); + resolve(value); + }, + (error) => { + release(); + reject(error); + }, + ); + }; + if (active < concurrency) { + run(); + } else { + queue.push(run); + } + }); +}