From 0f6fcd226be80b07842073ce2c8b3b565d851572 Mon Sep 17 00:00:00 2001 From: Andrew Mikofalvy <5668128+amikofalvy@users.noreply.github.com> Date: Tue, 21 Jul 2026 21:15:38 -0700 Subject: [PATCH] perf(server): worker pool for markdown parse offload (#2828) * perf(server): offload bridge-intake markdown parse to a worker pool * refactor(server): hoist fm-patch parse guess into agent-sessions helper * test(bridge): record S2 pre-merge fuzz and stress gate measurements GitOrigin-RevId: 397aee50d59a6c46e0016b0d7a79c4436bc0bbb0 --- .changeset/parse-pool-offload.md | 5 + knip.config.ts | 4 +- packages/cli/src/parse-worker.ts | 9 + packages/cli/tsdown.config.ts | 5 +- packages/server/package.json | 5 + packages/server/src/agent-sessions.ts | 235 +++++++++---- packages/server/src/api-extension.ts | 102 +++++- packages/server/src/bridge-intake.ts | 50 ++- packages/server/src/parse-pool.test.ts | 355 +++++++++++++++++++ packages/server/src/parse-pool.ts | 464 +++++++++++++++++++++++++ packages/server/src/parse-worker.ts | 110 ++++++ packages/server/src/server-factory.ts | 17 + packages/server/tsdown.config.ts | 6 +- 13 files changed, 1276 insertions(+), 91 deletions(-) create mode 100644 .changeset/parse-pool-offload.md create mode 100644 packages/cli/src/parse-worker.ts create mode 100644 packages/server/src/parse-pool.test.ts create mode 100644 packages/server/src/parse-pool.ts create mode 100644 packages/server/src/parse-worker.ts diff --git a/.changeset/parse-pool-offload.md b/.changeset/parse-pool-offload.md new file mode 100644 index 000000000..d535f3139 --- /dev/null +++ b/.changeset/parse-pool-offload.md @@ -0,0 +1,5 @@ +--- +"@inkeep/open-knowledge": patch +--- + +Agent writes no longer block the server on large-document markdown parsing. The parse that turns a written document's markdown into editor structure now runs on a bounded worker-thread pool for documents past 8KB, so a big write (or rollback, edit, frontmatter patch) no longer freezes every other in-flight agent tool call while it parses — the event-loop stall for a concurrent 1MB write drops from roughly 2 seconds to roughly half a second, and small-write latency under load matches idle latency. Small documents keep the faster inline path, output is byte-identical either way, and any worker failure transparently falls back to the previous inline behavior. diff --git a/knip.config.ts b/knip.config.ts index db361694e..296751a02 100644 --- a/knip.config.ts +++ b/knip.config.ts @@ -119,12 +119,12 @@ export default { entry: ['src/**/*.test.{ts,tsx}'], }, 'packages/server': { - entry: ['src/**/*.test.ts'], + entry: ['src/**/*.test.ts', 'src/parse-worker.ts'], project: 'src/**', ignoreDependencies: ['@types/shell-quote'], }, 'packages/cli': { - entry: ['src/**/*.test.ts', 'scripts/*.ts', 'tests/**/*.ts'], + entry: ['src/**/*.test.ts', 'scripts/*.ts', 'tests/**/*.ts', 'src/parse-worker.ts'], ignoreDependencies: [ '@inkeep/open-knowledge-app', // the CLI's `build:assets` script runs `cp -r ../app/dist dist/public` ], diff --git a/packages/cli/src/parse-worker.ts b/packages/cli/src/parse-worker.ts new file mode 100644 index 000000000..5cf464bfb --- /dev/null +++ b/packages/cli/src/parse-worker.ts @@ -0,0 +1,9 @@ +/** + * Worker-thread entry for the bundled CLI. The published package (and the + * packaged desktop app, which spawns dist/cli.mjs) ships no node_modules, + * so the parse pool's sibling probe (`./parse-worker.mjs` next to the + * importing bundle) is the only resolution path — this file exists solely + * to make tsdown emit that sibling. All logic lives in the server package; + * see `packages/server/src/parse-worker.ts`. + */ +import '@inkeep/open-knowledge-server/parse-worker'; diff --git a/packages/cli/tsdown.config.ts b/packages/cli/tsdown.config.ts index 1da1e82b2..ab701f934 100644 --- a/packages/cli/tsdown.config.ts +++ b/packages/cli/tsdown.config.ts @@ -28,7 +28,10 @@ console.warn = (...args: unknown[]) => { }; export default defineConfig({ - entry: { cli: 'src/cli.ts', index: 'src/index.ts' }, + // `parse-worker` ships as its own entry so the server's parse pool can + // spawn `./parse-worker.mjs` next to dist/cli.mjs at runtime (the + // published install has no node_modules to resolve through). + entry: { cli: 'src/cli.ts', index: 'src/index.ts', 'parse-worker': 'src/parse-worker.ts' }, unbundle: false, format: 'esm', dts: true, diff --git a/packages/server/package.json b/packages/server/package.json index 6b763efb2..aaa2143b5 100644 --- a/packages/server/package.json +++ b/packages/server/package.json @@ -9,6 +9,11 @@ "development": "./src/index.ts", "types": "./src/index.ts", "default": "./dist/index.mjs" + }, + "./parse-worker": { + "development": "./src/parse-worker.ts", + "types": "./src/parse-worker.ts", + "default": "./dist/parse-worker.mjs" } }, "files": [ diff --git a/packages/server/src/agent-sessions.ts b/packages/server/src/agent-sessions.ts index df0bac35c..3cbbbd021 100644 --- a/packages/server/src/agent-sessions.ts +++ b/packages/server/src/agent-sessions.ts @@ -19,6 +19,8 @@ */ import type { DirectConnection, Document, Hocuspocus } from '@hocuspocus/server'; import { + applyPatchToFm, + detectFmRegion, parseFrontmatterYaml, prependFrontmatter, stripFrontmatter, @@ -31,6 +33,7 @@ import * as Y from 'yjs'; import { composeAndWriteRawBody, deriveFragmentFromYtext, + type PrecomputedParse, replaceRawBody, } from './bridge-intake.ts'; import { isConfigDoc, isSystemDoc } from './cc1-broadcast.ts'; @@ -44,6 +47,7 @@ import { FrontmatterMalformedError } from './frontmatter-malformed-error.ts'; import { recordFrontmatterEditSurface } from './frontmatter-telemetry.ts'; import { getLogger } from './logger.ts'; import { incrementAgentSessionEvictions } from './metrics.ts'; +import { precomputeParse } from './parse-pool.ts'; import type { PairedWriteOrigin } from './server-observers.ts'; import { getMeter, setActiveSpanAttributes, withSpanSync } from './telemetry.ts'; @@ -141,6 +145,61 @@ function docNameToFile(docName: string): string { * @see PRECEDENTS.md precedent #11(a) (item-preserving cross-CRDT sync) * @see PRECEDENTS.md precedent #38 (Y.Text-is-truth contract) */ +/** + * Off-thread parse precompute for an agent write (see `parse-pool.ts`). + * Call BEFORE entering the write transact: composes the projected + * post-write bytes from the current Y.Text snapshot and parses them on a + * worker thread so a large-doc parse does not block the event loop. + * + * Purely advisory: returns `undefined` whenever the inline path should run + * (small doc, pool unavailable, conflict-gated doc, no-op composition), + * and the byte-identity guard in bridge-intake discards the result if the + * doc moved during the await — `applyAgentMarkdownWrite` recomposes inside + * the transact and stays the single source of applied bytes either way. + */ +export async function prepareAgentMarkdownParse( + document: Document, + markdown: string, + position: 'append' | 'prepend' | 'replace' | 'patch', + embedResolver?: { + resolveEmbed: (basename: string, sourcePath: string) => string | null; + sourcePath: string; + }, +): Promise { + // Conflict-gated docs refuse the write in apply; skip the wasted parse. + if (isDocInConflict(document)) return undefined; + const composed = composeAgentWrite(document.getText('source').toString(), markdown, position); + if (composed === undefined) return undefined; + return precomputeParse(composed.newContent, embedResolver); +} + +/** + * Off-thread parse precompute for the frontmatter-patch handler. Applies + * the FM patch to a PRE-transact snapshot and, when it would change the + * fenced region, parses the guessed full bytes on the worker pool. + * Although only the YAML region changes, the fragment still re-derives + * from the full post-patch body — that body parse is the cost this moves + * off-thread. The handler's in-transact `applyPatchToFm` splice stays + * authoritative (including the fence-separator rule mirrored here); the + * byte-identity guard in bridge-intake discards a stale guess. + * + * Lives here rather than in the handler so `api-extension.ts` keeps + * exactly one `applyPatchToFm(` call site per handler — the + * presence-pairing structural sweep counts those call sites as handler + * entries. + */ +export async function prepareFrontmatterPatchParse( + document: Document, + patch: Parameters[1], +): Promise { + const snapshot = document.getText('source').toString(); + const { fenced, body } = detectFmRegion(snapshot); + const result = applyPatchToFm(fenced, patch); + if (!result.ok || result.nextFenced === fenced) return undefined; + const needsFenceSeparator = fenced === '' && body !== '' && !body.startsWith('\n'); + return precomputeParse(result.nextFenced + (needsFenceSeparator ? '\n' : '') + body); +} + export function applyAgentMarkdownWrite( document: Document, markdown: string, @@ -155,6 +214,13 @@ export function applyAgentMarkdownWrite( resolveEmbed: (basename: string, sourcePath: string) => string | null; sourcePath: string; }, + /** + * Optional off-thread parse from `prepareAgentMarkdownParse`, computed + * against a pre-transact snapshot. Honored only when the recompose below + * produces byte-identical content (the guard lives in bridge-intake); a + * doc that moved during the caller's await falls back to inline parse. + */ + precomputed?: PrecomputedParse, ): AgentWriteContentDivergence | undefined { // Conflict-aware write gate (precedent #38 + this batch's structural // refusal contract). Lives OUTSIDE the transact — the check is static on @@ -176,7 +242,13 @@ export function applyAgentMarkdownWrite( }, }, () => { - const divergence = applyAgentMarkdownWriteInner(document, markdown, position, embedResolver); + const divergence = applyAgentMarkdownWriteInner( + document, + markdown, + position, + embedResolver, + precomputed, + ); if (divergence !== undefined) { setActiveSpanAttributes({ 'agent.content_divergent': true, @@ -208,6 +280,95 @@ export function snapshotBlocks(document: Document): string[] { .map((child) => child.toString()); } +/** + * The pure string-composition half of an agent write, shared by the apply + * path (inside the caller's transact) and `prepareAgentMarkdownParse` + * (before it). Splits the payload, applies the position semantics, and + * returns the exact full-document bytes the primitives will apply. + * Returns `undefined` for the append/prepend empty-body no-op. + * + * Split semantics: the agent may send a full document (FM + body) or + * body-only; we handle both. On 'replace', an FM in the payload supersedes + * the existing FM. On 'prepend'/'append', the payload's FM (if any) is + * dropped defensively to avoid producing a document with two FM blocks + * (double-FM is a CommonMark invalid state). Stripping FM is orthogonal to + * the byte-faithful contract — body bytes survive verbatim, only the + * FM-handling logic prevents structural breakage. + * + * Join semantics: `replace` keeps the payload verbatim (the byte-faithful + * primary path is untouched). For append/prepend, the agent's payload + * CONTENT still survives verbatim — only the join SEAM is normalized to + * exactly one blank-line separator. A prior payload's trailing newline + * stored in `currentBody` previously compounded with the `\n\n` separator + * into a `\n\n\n` double blank line: the existing body ended with `\n`, + * and the separator added two more. Trim trailing newlines off the leading + * chunk and leading newlines off the trailing chunk so the seam is always + * a single blank line, regardless of which side carried stray newlines. + * Within-payload blank lines and the far (non-seam) edge are untouched; + * empty-doc detection still uses `currentBody.length > 0`. + */ +interface ComposedAgentWrite { + existingFm: string; + finalFm: string; + /** Full document bytes (FM + body) the primitives apply verbatim. */ + newContent: string; +} + +function composeAgentWrite( + currentYText: string, + markdown: string, + position: 'append' | 'prepend' | 'replace' | 'patch', +): ComposedAgentWrite | undefined { + const { frontmatter: existingFm, body: currentBody } = stripFrontmatter(currentYText); + const { frontmatter: payloadFm, body: payloadBody } = stripFrontmatter(markdown); + + // append/prepend with an empty body is a no-op — there is nothing to add, + // so return before the write and leave the document byte-unchanged. + // Without this guard the `${payloadBody}\n\n${currentBody}` join below + // would inject a stray `\n\n` around empty content. A `---` + // frontmatter-only payload also lands here: append/prepend drop the + // payload FM, leaving an empty body — genuinely nothing to apply. + // Deliberate vertical whitespace remains expressible by sending it as + // body bytes (e.g. "\n"), which survive verbatim. + if ((position === 'append' || position === 'prepend') && payloadBody === '') { + return undefined; + } + + let finalFm: string; + let newBody: string; + switch (position) { + case 'replace': + finalFm = payloadFm || existingFm; + newBody = payloadBody; + break; + // `patch` (the edit find/replace path) composes identically to + // `replace` — full recomposed body, same FM supersede — but dispatches to + // the INCREMENTAL primitive (composeAndWriteRawBody), not the atomic + // replaceRawBody, so a surgical edit stays item-preserving instead of + // churning the whole doc. Same byte result, minimal CRDT delta. + case 'patch': + finalFm = payloadFm || existingFm; + newBody = payloadBody; + break; + case 'prepend': + finalFm = existingFm; + newBody = + currentBody.length > 0 + ? `${payloadBody.replace(/\n+$/, '')}\n\n${currentBody.replace(/^\n+/, '')}` + : payloadBody; + break; + case 'append': + finalFm = existingFm; + newBody = + currentBody.length > 0 + ? `${currentBody.replace(/\n+$/, '')}\n\n${payloadBody.replace(/^\n+/, '')}` + : payloadBody; + break; + } + + return { existingFm, finalFm, newContent: prependFrontmatter(finalFm, newBody) }; +} + function applyAgentMarkdownWriteInner( document: Document, markdown: string, @@ -216,75 +377,16 @@ function applyAgentMarkdownWriteInner( resolveEmbed: (basename: string, sourcePath: string) => string | null; sourcePath: string; }, + precomputed?: PrecomputedParse, ): AgentWriteContentDivergence | undefined { try { const ytext = document.getText('source'); const currentYText = ytext.toString(); - const { frontmatter: existingFm, body: currentBody } = stripFrontmatter(currentYText); - - // Split the agent's payload into frontmatter + body. The agent may send - // a full document (FM + body) or body-only; we handle both. On 'replace', - // an FM in the payload supersedes the existing FM. On 'prepend'/'append', - // the payload's FM (if any) is dropped defensively to avoid producing a - // document with two FM blocks (double-FM is a CommonMark invalid state). - // Stripping FM is orthogonal to the byte-faithful contract — body bytes - // survive verbatim, only the FM-handling logic prevents structural breakage. - const { frontmatter: payloadFm, body: payloadBody } = stripFrontmatter(markdown); - - // append/prepend with an empty body is a no-op — there is nothing to add, - // so return before the write and leave the document byte-unchanged. - // Without this guard the `${payloadBody}\n\n${currentBody}` join below - // would inject a stray `\n\n` around empty content. A `---` - // frontmatter-only payload also lands here: append/prepend drop the - // payload FM, leaving an empty body — genuinely nothing to apply. - // Deliberate vertical whitespace remains expressible by sending it as - // body bytes (e.g. "\n"), which survive verbatim. - if ((position === 'append' || position === 'prepend') && payloadBody === '') { + const composed = composeAgentWrite(currentYText, markdown, position); + if (composed === undefined) { return; } - - // Compose final FM and body. `replace` keeps the payload verbatim (the - // byte-faithful primary path is untouched). For append/prepend, the agent's - // payload CONTENT still survives verbatim — only the join SEAM is normalized - // to exactly one blank-line separator. A prior payload's trailing newline - // stored in `currentBody` previously compounded with the `\n\n` separator - // into a `\n\n\n` double blank line: the existing body ended - // with `\n`, and the separator added two more. Trim trailing newlines off - // the leading chunk and leading newlines off the trailing chunk so the seam - // is always a single blank line, regardless of which side carried stray - // newlines. Within-payload blank lines and the far (non-seam) edge are - // untouched; empty-doc detection still uses `currentBody.length > 0`. - let finalFm: string; - let newBody: string; - switch (position) { - case 'replace': - finalFm = payloadFm || existingFm; - newBody = payloadBody; - break; - // `patch` (the edit find/replace path) composes identically to - // `replace` — full recomposed body, same FM supersede — but dispatches to - // the INCREMENTAL primitive below (composeAndWriteRawBody), not the atomic - // replaceRawBody, so a surgical edit stays item-preserving instead of - // churning the whole doc. Same byte result, minimal CRDT delta. - case 'patch': - finalFm = payloadFm || existingFm; - newBody = payloadBody; - break; - case 'prepend': - finalFm = existingFm; - newBody = - currentBody.length > 0 - ? `${payloadBody.replace(/\n+$/, '')}\n\n${currentBody.replace(/^\n+/, '')}` - : payloadBody; - break; - case 'append': - finalFm = existingFm; - newBody = - currentBody.length > 0 - ? `${currentBody.replace(/\n+$/, '')}\n\n${payloadBody.replace(/^\n+/, '')}` - : payloadBody; - break; - } + const { existingFm, finalFm, newContent } = composed; if (finalFm !== existingFm) { // Refuse the write when the agent's payload introduces unparseable @@ -345,11 +447,10 @@ function applyAgentMarkdownWriteInner( // position (including `patch`, the edit surgical path) takes the // incremental primitive. Two primitives, two intents — see // `bridge-intake.ts` file header for the full contrast. - const newContent = prependFrontmatter(finalFm, newBody); if (position === 'replace') { - replaceRawBody(document, newContent, embedResolver); + replaceRawBody(document, newContent, embedResolver, precomputed); } else { - composeAndWriteRawBody(document, newContent, 'agent', embedResolver); + composeAndWriteRawBody(document, newContent, 'agent', embedResolver, precomputed); } // Site A content-divergence gate (shared predicate). Read Y.Text diff --git a/packages/server/src/api-extension.ts b/packages/server/src/api-extension.ts index 59a3aa5de..666b973c7 100644 --- a/packages/server/src/api-extension.ts +++ b/packages/server/src/api-extension.ts @@ -287,6 +287,8 @@ import { applyAgentMarkdownWrite, applyAgentUndo, iconFromClientName, + prepareAgentMarkdownParse, + prepareFrontmatterPatchParse, snapshotBlocks, } from './agent-sessions.ts'; import { type NormalizedSummary, normalizeSummary } from './agent-write-summary.ts'; @@ -471,7 +473,7 @@ import { isOrphanMode, } from './backlink-index.ts'; import { getBootTimings } from './boot-timings.ts'; -import { composeAndWriteRawBody, replaceRawBody } from './bridge-intake.ts'; +import { composeAndWriteRawBody, type PrecomputedParse, replaceRawBody } from './bridge-intake.ts'; import { isConfigDoc, isSystemDoc } from './cc1-broadcast.ts'; import { withHiddenWindowsConsole } from './child-process-windows-hide.ts'; import type { ResolveStrategy } from './conflict-storage.ts'; @@ -580,6 +582,7 @@ import { incrementSummariesProvided, incrementSummariesTruncated, } from './metrics.ts'; +import { precomputeParse } from './parse-pool.ts'; import { isWithinDir, toPosix } from './path-utils.ts'; import { deleteReconciledBase, @@ -5375,6 +5378,21 @@ export function createApiExtension(options: ApiExtensionOptions): Extension { options.resolveEmbed, ); + // Off-thread parse precompute (parse-pool): parse the projected + // post-write bytes on a worker BEFORE entering the transact so a + // large-doc parse does not block concurrent requests. Advisory — + // the byte-identity guard in bridge-intake discards it if the doc + // moved during this await. + const writeMdEmbedResolver = options.resolveEmbed + ? { resolveEmbed: options.resolveEmbed, sourcePath: resolvedDocName } + : undefined; + const writeMdPrecomputed = await prepareAgentMarkdownParse( + session.dc.document, + body.markdown, + position, + writeMdEmbedResolver, + ); + const timestamp = new Date().toISOString(); // Site A content-divergence captured from the in-transact gate. @@ -5406,9 +5424,8 @@ export function createApiExtension(options: ApiExtensionOptions): Extension { session.dc.document, body.markdown, position, - options.resolveEmbed - ? { resolveEmbed: options.resolveEmbed, sourcePath: resolvedDocName } - : undefined, + writeMdEmbedResolver, + writeMdPrecomputed, ); const changedBlocks = @@ -5808,6 +5825,18 @@ export function createApiExtension(options: ApiExtensionOptions): Extension { options.resolveEmbed, ); + // Off-thread parse precompute — same advisory pattern as the + // single-write handler (byte-identity guard in bridge-intake). + const entryEmbedResolver = options.resolveEmbed + ? { resolveEmbed: options.resolveEmbed, sourcePath: resolvedDocName } + : undefined; + const entryPrecomputed = await prepareAgentMarkdownParse( + session.dc.document, + entry.markdown, + entry.position ?? 'append', + entryEmbedResolver, + ); + let writeDivergence: AgentWriteContentDivergence | undefined; // Register one-shot observer BEFORE write transact so YTextEvent.delta is captured captureEffect(session.dc.document.getText('source'), agentId, colorSeed, clientName); @@ -5818,9 +5847,8 @@ export function createApiExtension(options: ApiExtensionOptions): Extension { session.dc.document, entry.markdown, entry.position ?? 'append', - options.resolveEmbed - ? { resolveEmbed: options.resolveEmbed, sourcePath: resolvedDocName } - : undefined, + entryEmbedResolver, + entryPrecomputed, ); const changedBlocks = @@ -6039,6 +6067,14 @@ export function createApiExtension(options: ApiExtensionOptions): Extension { options.resolveEmbed, ); + // Optimistic off-thread parse precompute (parse-pool): apply the FM + // patch to a PRE-transact snapshot and parse the guessed full bytes + // on a worker. The in-transact applyPatchToFm below stays + // authoritative; the byte-identity guard in bridge-intake discards + // a stale guess. No embed resolver here, mirroring the inline call + // below. + const fmPatchPrecomputed = await prepareFrontmatterPatchParse(session.dc.document, patch); + const timestamp = new Date().toISOString(); // `applyPatchToFm` is a total function returning FmEditResult — its @@ -6112,7 +6148,13 @@ export function createApiExtension(options: ApiExtensionOptions): Extension { currentFenced === '' && currentBody !== '' && !currentBody.startsWith('\n'); const newFull = result.nextFenced + (needsFenceSeparator ? '\n' : '') + currentBody; - composeAndWriteRawBody(session.dc.document, newFull, 'agent'); + composeAndWriteRawBody( + session.dc.document, + newFull, + 'agent', + undefined, + fmPatchPrecomputed, + ); recordFrontmatterEditSurface('mcp-write'); bodyMutated = true; } @@ -7464,6 +7506,38 @@ export function createApiExtension(options: ApiExtensionOptions): Extension { options.resolveEmbed, ); + // Optimistic off-thread parse precompute (parse-pool): splice the + // find/replace against a PRE-transact snapshot and parse the guessed + // post-patch bytes on a worker. The in-transact splice below stays + // the single authoritative compose — if the doc moved during this + // await the recomposed bytes differ and the byte-identity guard in + // bridge-intake discards the guess (inline parse, prior behavior). + const patchEmbedResolver = options.resolveEmbed + ? { resolveEmbed: options.resolveEmbed, sourcePath: docName } + : undefined; + let patchPrecomputed: PrecomputedParse | undefined; + { + const preSnapshot = session.dc.document.getText('source').toString(); + const { frontmatter: preFm, body: preBody } = stripFrontmatter(preSnapshot); + const preFull = prependFrontmatter(preFm, preBody); + const prePos = + offset == null + ? preFull.indexOf(find) + : preFull.slice(offset, offset + find.length) === find + ? offset + : -1; + if (prePos !== -1 && prePos >= preFm.length) { + const guessFull = + preFull.slice(0, prePos) + replace + preFull.slice(prePos + find.length); + patchPrecomputed = await prepareAgentMarkdownParse( + session.dc.document, + stripFrontmatter(guessFull).body, + 'patch', + patchEmbedResolver, + ); + } + } + const timestamp = new Date().toISOString(); let notFound = false; @@ -7563,9 +7637,8 @@ export function createApiExtension(options: ApiExtensionOptions): Extension { session.dc.document, newBody, 'patch', - options.resolveEmbed - ? { resolveEmbed: options.resolveEmbed, sourcePath: docName } - : undefined, + patchEmbedResolver, + patchPrecomputed, ); const changedBlocks = @@ -8886,6 +8959,11 @@ export function createApiExtension(options: ApiExtensionOptions): Extension { const rollbackEmbedResolver = options.resolveEmbed ? { resolveEmbed: options.resolveEmbed, sourcePath: docName } : undefined; + // Off-thread parse precompute: `markdown` (the target-version bytes) + // is fixed before this point, so unlike the compose-based writes the + // precompute can never go stale — the byte-identity guard always + // matches. A failed precompute degrades to the inline parse. + const rollbackPrecomputed = await precomputeParse(markdown, rollbackEmbedResolver); // Site A content-divergence gate for rollback — computed INSIDE the // transact, matching the write/patch gate. `ytext.toString()` here sees // `replaceRawBody`'s atomic post-state before observer settlement fires @@ -8896,7 +8974,7 @@ export function createApiExtension(options: ApiExtensionOptions): Extension { // `currentState` so the agent recovers without a re-read. let rollbackDivergence: AgentWriteContentDivergence | undefined; document.transact(() => { - replaceRawBody(document, markdown, rollbackEmbedResolver); + replaceRawBody(document, markdown, rollbackEmbedResolver, rollbackPrecomputed); rollbackDivergence = evaluateContentDivergence( document.getText('source').toString(), markdown, diff --git a/packages/server/src/bridge-intake.ts b/packages/server/src/bridge-intake.ts index 83f457c2e..406c2fb4f 100644 --- a/packages/server/src/bridge-intake.ts +++ b/packages/server/src/bridge-intake.ts @@ -45,6 +45,7 @@ * ytext bytes, silently reverting the write. */ import { applyFastDiff, stripFrontmatter } from '@inkeep/open-knowledge-core'; +import type { JSONContent } from '@tiptap/core'; import { updateYFragment } from '@tiptap/y-tiptap'; import type * as Y from 'yjs'; import { mdManager, schema } from './md-manager.ts'; @@ -77,6 +78,43 @@ interface EmbedResolverContext { */ type EmbedResolverArg = EmbedResolverContext | false | undefined; +/** + * Off-thread parse result a caller computed BEFORE entering its + * `doc.transact` (see `parse-pool.ts`). The primitives honor it only when + * `rawContent` byte-matches the bytes being applied in this call — the + * guard makes staleness structurally impossible: a doc that moved between + * the precompute and the transact fails the byte compare and the primitive + * parses inline exactly as it always has. Callers therefore never need to + * re-validate a precompute themselves. + */ +export interface PrecomputedParse { + /** Exact full-document bytes (frontmatter + body) the parse came from. */ + rawContent: string; + /** `mdManager.parseWithFallback(stripFrontmatter(rawContent).body, ...)` output. */ + parsedJson: JSONContent; +} + +/** + * Byte-guarded parse: use the precompute when it matches the bytes being + * applied, else parse inline (the pre-pool behavior, unchanged). + */ +function parseBodyWithPrecompute( + document: Y.Doc, + rawContent: string, + embedResolver: EmbedResolverArg, + precomputed: PrecomputedParse | undefined, +): JSONContent { + const { body } = stripFrontmatter(rawContent); + if (precomputed !== undefined && precomputed.rawContent === rawContent) { + return precomputed.parsedJson; + } + return withSpanSync( + 'md.parseWithFallback', + { attributes: { 'body.bytes': body.length, 'doc.name': document.guid } }, + () => mdManager.parseWithFallback(body, buildParseOpts(embedResolver)), + ); +} + function buildParseOpts(embedResolver: EmbedResolverArg): | { resolveEmbed: EmbedResolverContext['resolveEmbed']; @@ -148,6 +186,7 @@ export function composeAndWriteRawBody( rawContent: string, surface: ComposeWriteSurface, embedResolver?: EmbedResolverArg, + precomputed?: PrecomputedParse, ): void { withSpanSync( 'bridge.composeAndWriteRawBody', @@ -166,12 +205,7 @@ export function composeAndWriteRawBody( // Fragment derives from BODY (without FM): the markdown parser only handles // body markdown. FM lives in the YAML region of Y.Text directly — the // fragment side never carries it. - const { body } = stripFrontmatter(rawContent); - const parsedJson = withSpanSync( - 'md.parseWithFallback', - { attributes: { 'body.bytes': body.length, 'doc.name': document.guid } }, - () => mdManager.parseWithFallback(body, buildParseOpts(embedResolver)), - ); + const parsedJson = parseBodyWithPrecompute(document, rawContent, embedResolver, precomputed); const pmNode = schema.nodeFromJSON(parsedJson); // Y.Text gets the raw bytes FIRST, then fragment derives. The order matters: @@ -226,6 +260,7 @@ export function replaceRawBody( document: Y.Doc, rawContent: string, embedResolver?: EmbedResolverArg, + precomputed?: PrecomputedParse, ): void { withSpanSync( 'bridge.replaceRawBody', @@ -239,8 +274,7 @@ export function replaceRawBody( const xmlFragment = document.getXmlFragment('default'); const ytext = document.getText('source'); - const { body } = stripFrontmatter(rawContent); - const parsedJson = mdManager.parseWithFallback(body, buildParseOpts(embedResolver)); + const parsedJson = parseBodyWithPrecompute(document, rawContent, embedResolver, precomputed); const pmNode = schema.nodeFromJSON(parsedJson); const currentText = ytext.toString(); diff --git a/packages/server/src/parse-pool.test.ts b/packages/server/src/parse-pool.test.ts new file mode 100644 index 000000000..fdce44825 --- /dev/null +++ b/packages/server/src/parse-pool.test.ts @@ -0,0 +1,355 @@ +/** + * Parse-pool contract tests: worker parse output is byte-identical to the + * inline `mdManager.parseWithFallback` path, the byte-identity guard in + * bridge-intake makes stale precomputes structurally harmless, and every + * pool failure mode degrades to the inline path instead of a hung or + * wrong write. + * + * Equivalence comparisons use canonical JSON text, not deep-strict + * equality: ProseMirror's `toJSON()` emits `attrs` objects with a null + * prototype, which the worker's structured-clone transfer normalizes to + * `Object.prototype`. `schema.nodeFromJSON` reads properties only, so the + * prototype difference is unobservable downstream — canonical JSON is the + * "same PM JSON" claim these tests pin. + * + * The worker resolves `@inkeep/open-knowledge-core` through the `default` + * export condition (built dist); run a core build before these tests if + * core sources changed (turbo's `test` task depends on `^build`). + */ +import { readFileSync } from 'node:fs'; +import { dirname, resolve } from 'node:path'; +import { fileURLToPath, pathToFileURL } from 'node:url'; +import type { Document } from '@hocuspocus/server'; +import { stripFrontmatter } from '@inkeep/open-knowledge-core'; +import { yXmlFragmentToProseMirrorRootNode } from '@tiptap/y-tiptap'; +import { afterEach, describe, expect, test } from 'vitest'; +import * as Y from 'yjs'; +import { applyAgentMarkdownWrite, prepareAgentMarkdownParse } from './agent-sessions.ts'; +import { composeAndWriteRawBody, replaceRawBody } from './bridge-intake.ts'; +import { mdManager, schema } from './md-manager.ts'; +import { + _overrideParseTaskTimeoutForTests, + _overrideParseWorkerUrlForTests, + destroyParsePool, + offloadParse, + PARSE_OFFLOAD_MIN_BYTES, + precomputeParse, +} from './parse-pool.ts'; + +const __dirname = dirname(fileURLToPath(import.meta.url)); + +/** Frozen paired-write origin for the primitive-level tests. */ +const TEST_ORIGIN = { + source: 'local', + context: { origin: 'agent', paired: true }, +} as const; + +function asDocument(ydoc: Y.Doc, name = 'doc.md'): Document { + return { + name, + awareness: undefined, + getText: (n: string) => ydoc.getText(n), + getMap: (n: string) => ydoc.getMap(n), + getXmlFragment: (n: string) => ydoc.getXmlFragment(n), + transact: (fn: () => void, origin?: unknown) => ydoc.transact(fn, origin), + on: ydoc.on.bind(ydoc), + off: ydoc.off.bind(ydoc), + } as unknown as Document; +} + +function fragmentJson(ydoc: Y.Doc): string { + return JSON.stringify( + yXmlFragmentToProseMirrorRootNode(ydoc.getXmlFragment('default'), schema).toJSON(), + ); +} + +/** Deterministic feature-rich markdown large enough to cross the offload threshold. */ +function largeMarkdown(): string { + const section = [ + '## Heading with **strong** and _emphasis_ and `code`', + '', + 'A paragraph with a [link](https://example.com/a) and #tag and [[Wiki Page]].', + '', + '- item one', + '- item two', + ' - nested', + '', + '| A | B |', + '| --- | --- |', + '| 1 | 2 |', + '', + '```ts', + 'const x: number = 1;', + '```', + '', + '> quoted text', + '', + ].join('\n'); + return section.repeat(Math.ceil((PARSE_OFFLOAD_MIN_BYTES * 4) / section.length)); +} + +/** + * Byte-exercising fixtures for the worker-vs-inline equivalence sweep. + * Each hits a different pipeline surface: fallback recovery, PUA + * sentinels, hard-break trimming (the patched dependency), math, tables, + * escapes, raw HTML, JSX, reference links. + */ +const EQUIVALENCE_FIXTURES: ReadonlyArray<[string, string]> = [ + ['plain paragraph', 'Just a paragraph.\n'], + ['heading and setext', '# H1\n\nSetext\n===\n\nBody.\n'], + ['emphasis nesting', '**bold _nested_ and `code`** tail\n'], + ['escapes survive', 'not \\*emphasis\\* and a literal \\_underscore\\_\n'], + ['hard break then spaced text (patched dependency)', 'foo\\\n *bar*\n'], + ['hard break then whitespace-only text (patched dependency)', 'a\\\n \nb\n'], + ['task list', '- [ ] open\n- [x] done\n'], + ['ordered list renumber source', '3. three\n4. four\n'], + ['table alignment', '| a | b |\n|:--|--:|\n| 1 | 2 |\n'], + ['fenced code with lang', '```python\nprint("hi")\n```\n'], + ['math inline and block', 'Euler: $e^{i\\pi}+1=0$\n\n$$\nx^2\n$$\n'], + ['wikilink and tag', 'See [[Other Page|alias]] and #topic\n'], + ['raw inline html', 'line
break and span\n'], + ['jsx component', 'Body text\n'], + ['broken mdx falls back', 'before\n\n\ntext\n\n'], + ['reference link', 'See [ref one][r1].\n\n[r1]: https://example.com/r1\n'], + ['thematic break and blockquote', '---\n\n> quote\n\n---\n'], + ['crlf-free multi-blank', 'a\n\n\n\nb\n'], +]; + +afterEach(async () => { + _overrideParseWorkerUrlForTests(undefined); + _overrideParseTaskTimeoutForTests(undefined); + await destroyParsePool(); +}); + +describe('worker parse equivalence', () => { + test('fixture corpus: worker output is byte-identical to inline parse', async () => { + for (const [label, fixture] of EQUIVALENCE_FIXTURES) { + const inline = mdManager.parseWithFallback(fixture); + const offloaded = await offloadParse(fixture); + expect(JSON.stringify(offloaded), label).toBe(JSON.stringify(inline)); + } + }, 60_000); + + test('large generated doc: worker output is byte-identical to inline parse', async () => { + const md = largeMarkdown(); + const inline = mdManager.parseWithFallback(md); + const offloaded = await offloadParse(md); + expect(JSON.stringify(offloaded)).toBe(JSON.stringify(inline)); + }, 60_000); + + test('this file round-trips identically (real-world prose corpus)', async () => { + const source = readFileSync(resolve(__dirname, 'parse-pool.test.ts'), 'utf8'); + const asMarkdown = `# Source dump\n\n\`\`\`ts\n${source}\n\`\`\`\n`; + const inline = mdManager.parseWithFallback(asMarkdown); + const offloaded = await offloadParse(asMarkdown); + expect(JSON.stringify(offloaded)).toBe(JSON.stringify(inline)); + }, 60_000); + + test('patched dependency behavior holds inside the worker', async () => { + // The pinned @handlewithcare/remark-prosemirror patch drops a + // whitespace-only text node after a hard break instead of throwing + // RangeError from schema.text(''). An unpatched worker install would + // recover through the whole-doc fallback (a single paragraph of raw + // text) — assert the structured shape positively so the test fails + // even if BOTH sides were unpatched. + const offloaded = await offloadParse('foo\\\n *bar*'); + const paragraph = offloaded.content?.[0]; + expect(paragraph?.type).toBe('paragraph'); + const types = (paragraph?.content ?? []).map((n) => n.type); + expect(types).toContain('hardBreak'); + const barNode = (paragraph?.content ?? []).find((n) => n.type === 'text' && n.text === 'bar'); + expect(barNode?.marks?.some((m) => m.type === 'emphasis')).toBe(true); + }, 60_000); + + test('wiki-embed resolution matches inline (two-pass table protocol)', async () => { + const md = 'Intro paragraph.\n\n![[photo.png]]\n\n![[missing.bin]]\n'; + const resolver = { + resolveEmbed: (target: string) => (target === 'photo.png' ? 'assets/photo.png' : null), + resolveSize: (target: string) => (target === 'photo.png' ? 12_345 : null), + sourcePath: 'docs/page', + }; + const inline = mdManager.parseWithFallback(md, { ...resolver }); + const offloaded = await offloadParse(md, resolver); + expect(JSON.stringify(offloaded)).toBe(JSON.stringify(inline)); + }, 60_000); + + test('embed-free doc with a resolver present completes in one pass and matches inline', async () => { + const md = 'No embeds here, just a [link](https://example.com).\n'; + const resolver = { + resolveEmbed: () => { + throw new Error('resolver must not be consulted for an embed-free doc'); + }, + sourcePath: 'docs/page', + }; + const inline = mdManager.parseWithFallback(md, { + sourcePath: 'docs/page', + resolveEmbed: () => null, + }); + const offloaded = await offloadParse(md, resolver); + expect(JSON.stringify(offloaded)).toBe(JSON.stringify(inline)); + }, 60_000); +}); + +describe('precomputeParse threshold and fallback', () => { + test('small docs stay inline (returns undefined)', async () => { + const result = await precomputeParse('# tiny\n\nbody\n'); + expect(result).toBeUndefined(); + }); + + test('large docs offload and carry the exact rawContent', async () => { + const raw = `---\ntitle: X\n---\n\n${largeMarkdown()}`; + const result = await precomputeParse(raw); + expect(result).toBeDefined(); + expect(result?.rawContent).toBe(raw); + const inline = mdManager.parseWithFallback(stripFrontmatter(raw).body); + expect(JSON.stringify(result?.parsedJson)).toBe(JSON.stringify(inline)); + }, 60_000); + + test('worker file unavailable degrades to undefined (inline fallback)', async () => { + _overrideParseWorkerUrlForTests(null); + const result = await precomputeParse(largeMarkdown()); + expect(result).toBeUndefined(); + }); + + test('worker spawn failure degrades to undefined (inline fallback)', async () => { + _overrideParseWorkerUrlForTests(pathToFileURL(resolve(__dirname, 'no-such-worker.mjs'))); + const result = await precomputeParse(largeMarkdown()); + expect(result).toBeUndefined(); + }, 60_000); + + test('task timeout degrades to undefined, then the pool recovers', async () => { + _overrideParseTaskTimeoutForTests(1); + const timedOut = await precomputeParse(largeMarkdown()); + expect(timedOut).toBeUndefined(); + _overrideParseTaskTimeoutForTests(undefined); + const recovered = await precomputeParse(largeMarkdown()); + expect(recovered).toBeDefined(); + }, 60_000); + + test('destroyParsePool terminates workers and the next dispatch respawns', async () => { + const before = await precomputeParse(largeMarkdown()); + expect(before).toBeDefined(); + await destroyParsePool(); + const after = await precomputeParse(largeMarkdown()); + expect(after).toBeDefined(); + }, 60_000); +}); + +describe('bridge-intake byte-identity guard', () => { + test('a stale precompute is discarded (inline parse applies the real bytes)', () => { + const raw = '# Real\n\nreal body\n'; + const staleParse = mdManager.parseWithFallback('# Impostor\n\nimpostor body\n'); + const withStale = new Y.Doc(); + withStale.transact(() => { + composeAndWriteRawBody(withStale, raw, 'agent', undefined, { + rawContent: '# Impostor\n\nimpostor body\n', + parsedJson: staleParse, + }); + }, TEST_ORIGIN); + const control = new Y.Doc(); + control.transact(() => { + composeAndWriteRawBody(control, raw, 'agent'); + }, TEST_ORIGIN); + expect(withStale.getText('source').toString()).toBe(raw); + expect(fragmentJson(withStale)).toBe(fragmentJson(control)); + }); + + test('a byte-matching precompute is honored (observable via a divergent parse)', () => { + // Production precomputes always hold a true parse of rawContent; the + // deliberately-divergent JSON here is the only way to OBSERVE which + // branch applied. A matching rawContent must use the supplied parse. + const raw = '# Real\n\nreal body\n'; + const divergent = mdManager.parseWithFallback('# Marker heading only\n'); + const doc = new Y.Doc(); + doc.transact(() => { + replaceRawBody(doc, raw, undefined, { rawContent: raw, parsedJson: divergent }); + }, TEST_ORIGIN); + // Y.Text still receives the raw bytes verbatim (precedent #38) — + // only the fragment derivation consumed the precompute. + expect(doc.getText('source').toString()).toBe(raw); + expect(fragmentJson(doc)).toContain('Marker heading only'); + }); +}); + +describe('prepareAgentMarkdownParse end-to-end', () => { + test('fresh precompute: applied write matches the inline-path control byte-for-byte', async () => { + const md = largeMarkdown(); + const prepared = new Y.Doc(); + const preparedDoc = asDocument(prepared); + const precomputed = await prepareAgentMarkdownParse(preparedDoc, md, 'replace'); + expect(precomputed).toBeDefined(); + prepared.transact(() => { + applyAgentMarkdownWrite(preparedDoc, md, 'replace', undefined, precomputed); + }, TEST_ORIGIN); + + const control = new Y.Doc(); + const controlDoc = asDocument(control); + control.transact(() => { + applyAgentMarkdownWrite(controlDoc, md, 'replace'); + }, TEST_ORIGIN); + + expect(prepared.getText('source').toString()).toBe(control.getText('source').toString()); + expect(fragmentJson(prepared)).toBe(fragmentJson(control)); + }, 60_000); + + test('doc moved during the await: stale precompute discarded, write still correct', async () => { + const md = largeMarkdown(); + const ydoc = new Y.Doc(); + const doc = asDocument(ydoc); + const precomputed = await prepareAgentMarkdownParse(doc, md, 'append'); + expect(precomputed).toBeDefined(); + // A concurrent writer lands between the precompute and the transact. + ydoc.transact(() => { + ydoc.getText('source').insert(0, '# Raced-in heading\n\n'); + }, TEST_ORIGIN); + ydoc.transact(() => { + applyAgentMarkdownWrite(doc, md, 'append', undefined, precomputed); + }, TEST_ORIGIN); + + const control = new Y.Doc(); + const controlDoc = asDocument(control); + control.transact(() => { + control.getText('source').insert(0, '# Raced-in heading\n\n'); + }, TEST_ORIGIN); + control.transact(() => { + applyAgentMarkdownWrite(controlDoc, md, 'append'); + }, TEST_ORIGIN); + + expect(ydoc.getText('source').toString()).toBe(control.getText('source').toString()); + expect(fragmentJson(ydoc)).toBe(fragmentJson(control)); + }, 60_000); + + test('no-op composition (empty append) returns undefined without dispatching', async () => { + const ydoc = new Y.Doc(); + const result = await prepareAgentMarkdownParse(asDocument(ydoc), '', 'append'); + expect(result).toBeUndefined(); + }); +}); + +describe('worker entry ships in every bundle shape', () => { + // Build-config source assertions (the runtime alternative is running two + // full tsdown builds per test run). Same justification as the sibling + // tsdown-bundle-coverage test: dist shape is decided entirely by these + // configs, and the parse pool's sibling probe depends on the entry name. + test('server tsdown config emits the parse-worker entry', () => { + const config = readFileSync(resolve(__dirname, '../tsdown.config.ts'), 'utf8'); + expect(config).toMatch(/'parse-worker':\s*'src\/parse-worker\.ts'/); + }); + + test('cli tsdown config emits the parse-worker entry next to dist/cli.mjs', () => { + const config = readFileSync(resolve(__dirname, '../../cli/tsdown.config.ts'), 'utf8'); + expect(config).toMatch(/'parse-worker':\s*'src\/parse-worker\.ts'/); + }); + + test('server package.json exports the parse-worker subpath for both conditions', () => { + const pkg = JSON.parse(readFileSync(resolve(__dirname, '../package.json'), 'utf8')) as { + exports: Record>; + }; + expect(pkg.exports['./parse-worker']).toEqual({ + development: './src/parse-worker.ts', + types: './src/parse-worker.ts', + default: './dist/parse-worker.mjs', + }); + }); +}); diff --git a/packages/server/src/parse-pool.ts b/packages/server/src/parse-pool.ts new file mode 100644 index 000000000..fa3abc409 --- /dev/null +++ b/packages/server/src/parse-pool.ts @@ -0,0 +1,464 @@ +/** + * Bounded worker_threads pool that moves the bridge-intake markdown parse + * off the server's single event-loop thread. + * + * Division of labor (the load-bearing boundary): workers do PURE COMPUTE — + * markdown body in, ProseMirror JSON out (`parse-worker.ts`). The main + * thread keeps every CRDT mutation inside the caller's + * `session.dc.document.transact(fn, session.origin)` and the three + * bridge-intake primitives. The pool's output is only ever consumed as a + * `PrecomputedParse` whose byte-identity guard lives in `bridge-intake.ts`: + * a precompute whose `rawContent` no longer matches the bytes being applied + * is silently discarded and the primitive parses inline, so a doc that + * moved during the `await` can never receive a stale fragment. + * + * Every failure mode (no worker file in this packaging, spawn failure, + * worker error/exit, task timeout, saturated queue) degrades to "return + * undefined" — the caller applies the write exactly as before this pool + * existed, parsing inline inside the transact. Offload is an optimization, + * never a correctness dependency. + * + * Worker-file resolution covers the three packaging shapes: + * 1. `./parse-worker.mjs` sibling — the server's own dist AND the + * published CLI bundle (`packages/cli` emits a `parse-worker` entry + * next to `dist/cli.mjs`; the packaged desktop app spawns that same + * bundle). + * 2. `./parse-worker.ts` sibling — source mode (vitest, tsx dev server). + * Node 24 runs the .ts entry via native type stripping; its imports + * resolve `@inkeep/open-knowledge-core` through the `default` export + * condition (the built core dist), so run a build after editing core + * or the worker parses stale code. + * 3. Package-resolved server dist — contexts that bundled the server + * into another app but still have node_modules (electron-vite desktop + * dev). When even that misses, the inline fallback engages. + */ +import { existsSync } from 'node:fs'; +import { createRequire } from 'node:module'; +import { availableParallelism } from 'node:os'; +import { dirname, join } from 'node:path'; +import { performance } from 'node:perf_hooks'; +import { fileURLToPath, pathToFileURL } from 'node:url'; +import { Worker } from 'node:worker_threads'; +import { stripFrontmatter } from '@inkeep/open-knowledge-core'; +import type { JSONContent } from '@tiptap/core'; +import type { PrecomputedParse } from './bridge-intake.ts'; +import { getLogger } from './logger.ts'; +import type { + ParseWorkerEmbedResolution, + ParseWorkerResult, + ParseWorkerTask, +} from './parse-worker.ts'; +import { getMeter, onTelemetryShutdown } from './telemetry.ts'; + +const log = getLogger('parse-pool'); + +/** + * Bodies below this size parse inline: at ~1ms/KB measured parse cost the + * sub-8KB range blocks the loop for under ~8ms, which is cheaper + * end-to-end than a worker round-trip and keeps small-write latency + * unchanged. + */ +export const PARSE_OFFLOAD_MIN_BYTES = 8 * 1024; + +/** + * Generous per-task ceiling: a legitimate multi-MB doc parses in seconds + * (~1.5s/MB measured), and `parseWithFallback`'s internal budget already + * bounds pathological fallback recursion. On expiry the task's worker is + * terminated (it may be wedged mid-parse) and the caller falls back inline. + */ +const PARSE_TASK_TIMEOUT_MS = 30_000; + +/** Reject dispatches beyond this backlog; callers parse inline instead. */ +const MAX_PENDING_TASKS = 32; + +/** Terminate workers idle this long; the pool respawns lazily on demand. */ +const WORKER_IDLE_REAP_MS = 30_000; + +const POOL_SIZE = Math.max(1, Math.min(4, availableParallelism() - 1)); + +/** Embed-resolver context accepted by `precomputeParse` (mirrors the + * bridge-intake `EmbedResolverContext` shape). */ +export interface ParsePoolEmbedResolver { + resolveEmbed: (basename: string, sourcePath: string) => string | null; + resolveSize?: (basename: string, sourcePath: string) => number | null; + sourcePath: string; +} + +type DispatchMode = + | 'offload' + | 'inline-small' + | 'inline-unavailable' + | 'inline-busy' + | 'inline-timeout' + | 'inline-error'; + +// ── Telemetry (bounded cardinality: mode is a 6-value enum) ───────── + +type Meter = ReturnType; +let dispatchCounter: ReturnType | null = null; +let taskLatencyHistogram: ReturnType | null = null; +let gaugesInstalled = false; + +onTelemetryShutdown(() => { + dispatchCounter = null; + taskLatencyHistogram = null; + gaugesInstalled = false; +}); + +function recordDispatch(mode: DispatchMode): void { + dispatchCounter ||= getMeter().createCounter('ok.parse_pool.dispatch_total', { + description: + 'Bridge-intake parse precompute dispatches by mode: offload (worker parse used) vs the inline-* fallback reasons (small doc, pool unavailable, queue saturated, task timeout, worker error).', + }); + dispatchCounter.add(1, { mode }); +} + +function recordTaskLatency(ms: number): void { + taskLatencyHistogram ||= getMeter().createHistogram('ok.parse_pool.task_ms', { + description: + 'Wall-clock latency of a completed parse-pool offload (both passes for embed-bearing docs), in milliseconds.', + unit: 'ms', + }); + taskLatencyHistogram.record(ms); +} + +function installGauges(): void { + if (gaugesInstalled) return; + gaugesInstalled = true; + getMeter() + .createObservableGauge('ok.parse_pool.queue_depth', { + description: 'Parse-pool tasks waiting for a free worker.', + }) + .addCallback((result) => { + result.observe(queue.length); + }); + getMeter() + .createObservableGauge('ok.parse_pool.workers', { + description: 'Live parse-pool worker threads.', + }) + .addCallback((result) => { + result.observe(workers.length); + }); +} + +// ── Pool state (module singleton; lazily spawned, lazily respawned) ── + +interface PendingTask { + task: ParseWorkerTask; + resolve: (result: ParseWorkerResult) => void; + reject: (err: Error) => void; +} + +interface PoolWorker { + worker: Worker; + current: PendingTask | null; + timer: NodeJS.Timeout | null; +} + +const workers: PoolWorker[] = []; +const queue: PendingTask[] = []; +let nextTaskId = 1; +let idleReapTimer: NodeJS.Timeout | null = null; +let workerUrlOverride: URL | null | undefined; + +/** Test hook: force the worker URL (or `null` = unavailable). `undefined` resets. */ +export function _overrideParseWorkerUrlForTests(url: URL | null | undefined): void { + workerUrlOverride = url; +} + +let taskTimeoutMs = PARSE_TASK_TIMEOUT_MS; + +/** Test hook: shrink the per-task timeout. `undefined` resets. */ +export function _overrideParseTaskTimeoutForTests(ms: number | undefined): void { + taskTimeoutMs = ms ?? PARSE_TASK_TIMEOUT_MS; +} + +function resolveWorkerUrl(): URL | null { + if (workerUrlOverride !== undefined) return workerUrlOverride; + const candidates: URL[] = [ + new URL('./parse-worker.mjs', import.meta.url), + new URL('./parse-worker.ts', import.meta.url), + ]; + try { + const requireFromHere = createRequire(import.meta.url); + candidates.push( + pathToFileURL( + join( + dirname(requireFromHere.resolve('@inkeep/open-knowledge-server/parse-worker')), + 'parse-worker.mjs', + ), + ), + ); + } catch { + // No resolvable node_modules (fully bundled install) — siblings only. + } + for (const candidate of candidates) { + try { + if (existsSync(fileURLToPath(candidate))) return candidate; + } catch { + // Non-file URL — skip. + } + } + return null; +} + +function spawnWorker(url: URL): PoolWorker | null { + try { + const worker = new Worker(url); + // An idle pool must never hold the process open: teardown paths that + // miss destroyParsePool() (crash handlers, test rigs) still exit. + worker.unref(); + const poolWorker: PoolWorker = { worker, current: null, timer: null }; + worker.on('message', (result: ParseWorkerResult) => { + completeTask(poolWorker, (pending) => pending.resolve(result)); + }); + worker.on('error', (err: Error) => { + completeTask(poolWorker, (pending) => pending.reject(err)); + removeWorker(poolWorker); + }); + worker.on('exit', () => { + completeTask(poolWorker, (pending) => + pending.reject(new Error('parse worker exited mid-task')), + ); + removeWorker(poolWorker); + }); + return poolWorker; + } catch (err) { + log.warn({ err }, '[parse-pool] failed to spawn parse worker'); + return null; + } +} + +function completeTask(poolWorker: PoolWorker, settle: (pending: PendingTask) => void): void { + const pending = poolWorker.current; + if (pending === null) return; + poolWorker.current = null; + if (poolWorker.timer !== null) { + clearTimeout(poolWorker.timer); + poolWorker.timer = null; + } + settle(pending); + pumpQueue(); +} + +function removeWorker(poolWorker: PoolWorker): void { + const idx = workers.indexOf(poolWorker); + if (idx !== -1) workers.splice(idx, 1); +} + +function pumpQueue(): void { + while (queue.length > 0) { + let idle = workers.find((w) => w.current === null); + if (idle === undefined && workers.length < POOL_SIZE) { + // A timed-out or crashed worker was removed while tasks were queued — + // respawn so the backlog drains instead of waiting out its timeout. + const url = resolveWorkerUrl(); + const spawned = url === null ? null : spawnWorker(url); + if (spawned !== null) { + workers.push(spawned); + idle = spawned; + } + } + if (idle === undefined) break; + const pending = queue.shift(); + if (pending === undefined) break; + assignTask(idle, pending); + } + scheduleIdleReap(); +} + +function assignTask(poolWorker: PoolWorker, pending: PendingTask): void { + poolWorker.current = pending; + poolWorker.timer = setTimeout(() => { + // The worker may be wedged inside a pathological parse — terminate it + // (the pool respawns lazily) and let the caller fall back inline. + poolWorker.current = null; + removeWorker(poolWorker); + void poolWorker.worker.terminate(); + pending.reject(new ParseTaskTimeoutError()); + pumpQueue(); + }, taskTimeoutMs); + poolWorker.timer.unref(); + poolWorker.worker.postMessage(pending.task); +} + +class ParseTaskTimeoutError extends Error { + constructor() { + super(`parse worker task exceeded ${taskTimeoutMs}ms`); + this.name = 'ParseTaskTimeoutError'; + } +} + +function scheduleIdleReap(): void { + if (idleReapTimer !== null) return; + if (workers.length === 0) return; + idleReapTimer = setTimeout(() => { + idleReapTimer = null; + const busy = workers.some((w) => w.current !== null); + if (busy || queue.length > 0) { + scheduleIdleReap(); + return; + } + for (const poolWorker of workers.splice(0)) { + void poolWorker.worker.terminate(); + } + }, WORKER_IDLE_REAP_MS); + idleReapTimer.unref(); +} + +function dispatch(task: Omit): Promise { + const url = resolveWorkerUrl(); + if (url === null) { + return Promise.reject(new ParsePoolUnavailableError()); + } + installGauges(); + const pending: PendingTask = { + task: { ...task, id: nextTaskId++ }, + resolve: () => {}, + reject: () => {}, + }; + const promise = new Promise((resolve, reject) => { + pending.resolve = resolve; + pending.reject = reject; + }); + const idle = workers.find((w) => w.current === null); + if (idle !== undefined) { + assignTask(idle, pending); + } else if (workers.length < POOL_SIZE) { + const spawned = spawnWorker(url); + if (spawned === null) return Promise.reject(new ParsePoolUnavailableError()); + workers.push(spawned); + assignTask(spawned, pending); + } else if (queue.length < MAX_PENDING_TASKS) { + queue.push(pending); + } else { + return Promise.reject(new ParsePoolBusyError()); + } + return promise; +} + +class ParsePoolUnavailableError extends Error { + constructor() { + super('parse worker unavailable in this packaging'); + this.name = 'ParsePoolUnavailableError'; + } +} + +class ParsePoolBusyError extends Error { + constructor() { + super(`parse pool backlog exceeded ${MAX_PENDING_TASKS} tasks`); + this.name = 'ParsePoolBusyError'; + } +} + +/** + * Terminate every worker and reject the backlog. NOT a permanent shutdown: + * the next `precomputeParse` respawns lazily, so multiple server instances + * in one process (test rigs, dev-server restarts) can each tear down + * without wedging the others — a rejected in-flight task simply falls back + * to the inline parse. + */ +export async function destroyParsePool(): Promise { + if (idleReapTimer !== null) { + clearTimeout(idleReapTimer); + idleReapTimer = null; + } + for (const pending of queue.splice(0)) { + pending.reject(new Error('parse pool destroyed')); + } + const terminating = workers.splice(0).map((poolWorker) => { + if (poolWorker.timer !== null) clearTimeout(poolWorker.timer); + const pending = poolWorker.current; + poolWorker.current = null; + pending?.reject(new Error('parse pool destroyed')); + return poolWorker.worker.terminate(); + }); + await Promise.allSettled(terminating); +} + +/** + * Offload one parse to the pool, throwing on any pool-level failure. + * Exported for the equivalence tests (loud failures beat silent fallback + * there); production callers use `precomputeParse`. + */ +export async function offloadParse( + body: string, + embedResolver?: ParsePoolEmbedResolver, +): Promise { + const base: Omit = { + body, + ...(embedResolver !== undefined + ? { + sourcePath: embedResolver.sourcePath, + recordEmbeds: true, + wantSizes: embedResolver.resolveSize !== undefined, + } + : {}), + }; + const first = await dispatch(base); + if (!first.ok) throw new Error(first.message); + if (first.requestedTargets === undefined || first.requestedTargets.length === 0) { + return first.parsedJson; + } + // Pass 2: the parser asked for embed targets the worker cannot resolve + // (fs + basename index are main-thread state). Resolve them here and + // re-parse with the table so the output matches inline byte-for-byte. + const resolver = embedResolver as ParsePoolEmbedResolver; + const embedTable: Record = {}; + for (const target of first.requestedTargets) { + embedTable[target] = { + path: resolver.resolveEmbed(target, resolver.sourcePath) ?? null, + size: resolver.resolveSize?.(target, resolver.sourcePath) ?? null, + }; + } + const second = await dispatch({ + body, + sourcePath: resolver.sourcePath, + wantSizes: resolver.resolveSize !== undefined, + embedTable, + }); + if (!second.ok) throw new Error(second.message); + return second.parsedJson; +} + +/** + * Precompute the bridge-intake parse for `rawContent` (full doc bytes, + * frontmatter + body) off-thread. Returns `undefined` whenever the inline + * path should run instead — small doc, pool unavailable/saturated, worker + * error or timeout. Never throws. + * + * The returned `PrecomputedParse` is only honored by the bridge-intake + * primitives when its `rawContent` byte-matches the bytes being applied, + * so callers may compute it from a pre-transact snapshot without any + * staleness risk. + */ +export async function precomputeParse( + rawContent: string, + embedResolver?: ParsePoolEmbedResolver, +): Promise { + const { body } = stripFrontmatter(rawContent); + if (body.length < PARSE_OFFLOAD_MIN_BYTES) { + recordDispatch('inline-small'); + return undefined; + } + const started = performance.now(); + try { + const parsedJson = await offloadParse(body, embedResolver); + recordDispatch('offload'); + recordTaskLatency(performance.now() - started); + return { rawContent, parsedJson }; + } catch (err) { + if (err instanceof ParsePoolUnavailableError) { + recordDispatch('inline-unavailable'); + } else if (err instanceof ParsePoolBusyError) { + recordDispatch('inline-busy'); + } else if (err instanceof ParseTaskTimeoutError) { + recordDispatch('inline-timeout'); + log.warn({ err, bodyBytes: body.length }, '[parse-pool] offload timed out; parsing inline'); + } else { + recordDispatch('inline-error'); + log.warn({ err, bodyBytes: body.length }, '[parse-pool] offload failed; parsing inline'); + } + return undefined; + } +} diff --git a/packages/server/src/parse-worker.ts b/packages/server/src/parse-worker.ts new file mode 100644 index 000000000..1c27aeb54 --- /dev/null +++ b/packages/server/src/parse-worker.ts @@ -0,0 +1,110 @@ +/** + * Markdown parse worker — the compute half of the bridge-intake parse + * offload (`parse-pool.ts`). + * + * PURE COMPUTE ONLY: markdown body bytes in, ProseMirror JSON out. No + * Y.js type ever crosses this boundary (Y.Doc / Y.Text are not + * structured-clonable), no fs access, no session or document state. All + * CRDT mutation stays on the main thread inside the caller's + * `doc.transact(..., origin)` and the bridge-intake primitives. + * + * Embed resolution (`![[photo.png]]` → disk path) needs the server's + * basename index and fs, so it cannot run here. The two-pass protocol + * keeps it exact without duplicating resolver logic: pass 1 parses with a + * recording resolver and reports which targets the parser actually asked + * for; the pool resolves those on the main thread and dispatches pass 2 + * with a plain-object table. A doc with no wiki-embeds (the common case) + * completes in pass 1 because the recording resolver is never invoked. + * + * Output is byte-identical to an inline `mdManager.parseWithFallback` + * call: both sides load the same `md-manager.ts` configuration (same + * `sharedExtensions`, same pinned+patched markdown dependencies out of + * the same install). `parse-pool.test.ts` pins the equivalence. + */ +import { parentPort } from 'node:worker_threads'; +import type { JSONContent } from '@tiptap/core'; +import { mdManager } from './md-manager.ts'; + +/** One embed target's main-thread resolution, shipped to pass 2. */ +export interface ParseWorkerEmbedResolution { + path: string | null; + size: number | null; +} + +export interface ParseWorkerTask { + id: number; + /** Frontmatter-free markdown body (callers strip the YAML region first). */ + body: string; + /** docName threaded to the embed resolvers for shortest-path computation. */ + sourcePath?: string; + /** Pass 1: install a recording resolver (caller has a live resolver). */ + recordEmbeds?: boolean; + /** + * Mirror the presence of the caller's `resolveSize`. The parse handlers + * key behavior off resolver PRESENCE, so the worker must not install a + * size resolver the inline path would not have had. + */ + wantSizes?: boolean; + /** Pass 2: resolved embed table keyed by the pass-1 recorded targets. */ + embedTable?: Record; +} + +export type ParseWorkerResult = + | { id: number; ok: true; parsedJson: JSONContent; requestedTargets?: string[] } + | { id: number; ok: false; message: string }; + +function runTask(task: ParseWorkerTask): ParseWorkerResult { + try { + let requested: Set | undefined; + let opts: + | { + sourcePath: string; + resolveEmbed: (target: string, sourcePath: string) => string | null; + resolveSize?: (target: string, sourcePath: string) => number | null; + } + | undefined; + if (task.embedTable !== undefined && task.sourcePath !== undefined) { + const table = task.embedTable; + opts = { + sourcePath: task.sourcePath, + resolveEmbed: (target) => table[target]?.path ?? null, + ...(task.wantSizes ? { resolveSize: (target) => table[target]?.size ?? null } : {}), + }; + } else if (task.recordEmbeds && task.sourcePath !== undefined) { + const record = new Set(); + requested = record; + opts = { + sourcePath: task.sourcePath, + resolveEmbed: (target) => { + record.add(target); + return null; + }, + ...(task.wantSizes + ? { + resolveSize: (target: string) => { + record.add(target); + return null; + }, + } + : {}), + }; + } + const parsedJson = mdManager.parseWithFallback(task.body, opts); + return requested !== undefined && requested.size > 0 + ? { id: task.id, ok: true, parsedJson, requestedTargets: [...requested] } + : { id: task.id, ok: true, parsedJson }; + } catch (err) { + // parseWithFallback never throws by contract; this catch is the wire- + // level backstop so a worker bug degrades to the pool's inline + // fallback instead of an unhandled worker crash. + return { + id: task.id, + ok: false, + message: err instanceof Error ? err.message.slice(0, 500) : String(err).slice(0, 500), + }; + } +} + +parentPort?.on('message', (task: ParseWorkerTask) => { + parentPort?.postMessage(runTask(task)); +}); diff --git a/packages/server/src/server-factory.ts b/packages/server/src/server-factory.ts index 9447a92b3..b3f7b4a38 100644 --- a/packages/server/src/server-factory.ts +++ b/packages/server/src/server-factory.ts @@ -138,6 +138,7 @@ import { incrementUpstreamImport, setRecentlyRemovedDocsSize, } from './metrics.ts'; +import { destroyParsePool } from './parse-pool.ts'; import { isWithinDir, toPosix } from './path-utils.ts'; import { createPersistenceExtension, @@ -2765,6 +2766,22 @@ export function createServer(options: ServerOptions): ServerInstance { log.error({ err }, '[server] shutdown phase-2 agent session drain failed'); } + // Phase 2b: parse-pool teardown rides the session drain — with + // sessions closed no new precompute dispatches from this server. + // destroyParsePool is reset-not-shutdown: another live server in + // the same process (test rigs, dev restarts) respawns workers + // lazily on its next write, and any in-flight task it loses falls + // back to the inline parse. + try { + await destroyParsePool(); + } catch (err) { + phaseErrors.push({ + phase: 'parse-pool-teardown', + error: err instanceof Error ? err.message : String(err), + }); + log.error({ err }, '[server] shutdown phase-2b parse pool teardown failed'); + } + // Phase 3: drain L1 (Y.Doc → markdown → disk) via afterUnloadDocument hook try { await flushAllStoresAndWait(destroyTimeoutMs); diff --git a/packages/server/tsdown.config.ts b/packages/server/tsdown.config.ts index fdd52b903..602bf5d69 100644 --- a/packages/server/tsdown.config.ts +++ b/packages/server/tsdown.config.ts @@ -1,7 +1,11 @@ import { defineConfig } from 'tsdown'; export default defineConfig({ - entry: { index: 'src/index.ts' }, + // `parse-worker` must stay its own entry: parse-pool.ts spawns it as a + // worker_threads file next to the importing bundle (`./parse-worker.mjs` + // sibling probe), so inlining it into index.mjs would leave no file to + // spawn in dist-based installs. + entry: { index: 'src/index.ts', 'parse-worker': 'src/parse-worker.ts' }, unbundle: false, format: 'esm', dts: false,