diff --git a/knowledge-fs/apps/api/package.json b/knowledge-fs/apps/api/package.json index 297c38308e3..827d81b9b5a 100644 --- a/knowledge-fs/apps/api/package.json +++ b/knowledge-fs/apps/api/package.json @@ -22,7 +22,8 @@ "@knowledge/generation": "workspace:*", "@knowledge/parsers": "workspace:*", "hono": "^4.12.25", - "sharp": "^0.35.3" + "sharp": "^0.35.3", + "undici": "^8.10.0" }, "devDependencies": { "@types/node": "^22.10.2", diff --git a/knowledge-fs/apps/api/src/parser-options.test.ts b/knowledge-fs/apps/api/src/parser-options.test.ts index 667d517679b..2245bb919c9 100644 --- a/knowledge-fs/apps/api/src/parser-options.test.ts +++ b/knowledge-fs/apps/api/src/parser-options.test.ts @@ -1,10 +1,51 @@ import { describe, expect, it } from "vitest"; -import { createApiDocumentParser } from "./parser-options"; +import type { Dispatcher } from "undici"; + +import { createApiDocumentParser, createNodeUnstructuredFetch } from "./parser-options"; const encoder = new TextEncoder(); describe("createApiDocumentParser", () => { + it("translates native requests onto a matching Node transport with aligned timeouts", async () => { + let dispatcherOptions: Readonly<{ bodyTimeout: number; headersTimeout: number }> | undefined; + let requestDispatcher: Dispatcher | undefined; + let receivedBody: BodyInit | null | undefined; + let receivedInput: RequestInfo | URL | undefined; + let receivedMethod: string | undefined; + const dispatcher = {} as Dispatcher; + const nodeFetch = createNodeUnstructuredFetch({ + createDispatcher: (options) => { + dispatcherOptions = options; + return dispatcher; + }, + fetch: async (input, init) => { + receivedBody = init?.body; + receivedInput = input; + receivedMethod = init?.method; + requestDispatcher = (init as (RequestInit & { dispatcher?: Dispatcher }) | undefined) + ?.dispatcher; + return new Response("[]"); + }, + requestTimeoutMs: 600_000, + }); + const request = new Request("https://unstructured.example.test/general/v0/general", { + body: "document", + method: "POST", + }); + + await nodeFetch(request); + + expect(dispatcherOptions).toEqual({ + bodyTimeout: 600_000, + headersTimeout: 600_000, + }); + expect(receivedInput).toBe(request.url); + expect(receivedMethod).toBe("POST"); + expect(receivedBody).toBe(request.body); + expect(requestDispatcher).toBe(dispatcher); + }); + it("keeps Markdown and structured data on native parsers", async () => { let fetchCalls = 0; const parser = createApiDocumentParser({ @@ -162,8 +203,11 @@ describe("createApiDocumentParser", () => { ).toThrow("UNSTRUCTURED_MAX_CONCURRENCY must be between 1 and 32"); expect(() => createApiDocumentParser({ - env: { UNSTRUCTURED_API_URL: "http://parser", UNSTRUCTURED_REQUEST_TIMEOUT_MS: "600001" }, + env: { + UNSTRUCTURED_API_URL: "http://parser", + UNSTRUCTURED_REQUEST_TIMEOUT_MS: "1800001", + }, }), - ).toThrow("UNSTRUCTURED_REQUEST_TIMEOUT_MS must be between 1 and 600000"); + ).toThrow("UNSTRUCTURED_REQUEST_TIMEOUT_MS must be between 1 and 1800000"); }); }); diff --git a/knowledge-fs/apps/api/src/parser-options.ts b/knowledge-fs/apps/api/src/parser-options.ts index 87a773424a8..2efdf41c55b 100644 --- a/knowledge-fs/apps/api/src/parser-options.ts +++ b/knowledge-fs/apps/api/src/parser-options.ts @@ -6,6 +6,23 @@ import { createParserRouter, createUnstructuredParserClient, } from "@knowledge/parsers"; +import { Agent, type Dispatcher, fetch as undiciFetch } from "undici"; + +const defaultUnstructuredRequestTimeoutMs = 120_000; +const maxUnstructuredRequestTimeoutMs = 1_800_000; + +interface NodeUnstructuredFetchOptions { + readonly createDispatcher?: ( + options: Readonly<{ bodyTimeout: number; headersTimeout: number }>, + ) => Dispatcher; + readonly fetch?: typeof fetch; + readonly requestTimeoutMs: number; +} + +type DispatcherRequestInit = RequestInit & { + readonly dispatcher: Dispatcher; + readonly duplex?: "half"; +}; export interface ApiParserEnv { readonly NODE_ENV?: string | undefined; @@ -60,13 +77,26 @@ function createApiUnstructuredParser({ }; } + const requestTimeoutMs = + env.UNSTRUCTURED_REQUEST_TIMEOUT_MS === undefined + ? defaultUnstructuredRequestTimeoutMs + : parseBoundedPositiveInteger( + env.UNSTRUCTURED_REQUEST_TIMEOUT_MS, + "UNSTRUCTURED_REQUEST_TIMEOUT_MS", + maxUnstructuredRequestTimeoutMs, + ); + return createUnstructuredParserClient({ ...(env.UNSTRUCTURED_DEFAULT_LANGUAGE?.trim() ? { defaultLanguage: env.UNSTRUCTURED_DEFAULT_LANGUAGE.trim() } : {}), endpoint, ...(env.UNSTRUCTURED_API_KEY?.trim() ? { apiKey: env.UNSTRUCTURED_API_KEY.trim() } : {}), - ...(fetchImpl ? { fetch: fetchImpl } : {}), + fetch: + fetchImpl ?? + createNodeUnstructuredFetch({ + requestTimeoutMs, + }), ...(env.UNSTRUCTURED_MAX_CONCURRENCY !== undefined ? { maxConcurrency: parseBoundedPositiveInteger( @@ -100,18 +130,42 @@ function createApiUnstructuredParser({ ), } : {}), - ...(env.UNSTRUCTURED_REQUEST_TIMEOUT_MS !== undefined - ? { - requestTimeoutMs: parseBoundedPositiveInteger( - env.UNSTRUCTURED_REQUEST_TIMEOUT_MS, - "UNSTRUCTURED_REQUEST_TIMEOUT_MS", - 600_000, - ), - } - : {}), + requestTimeoutMs, }); } +export function createNodeUnstructuredFetch({ + createDispatcher = (options) => new Agent(options), + fetch: fetchImpl = undiciFetch as unknown as typeof fetch, + requestTimeoutMs, +}: NodeUnstructuredFetchOptions): typeof fetch { + const dispatcher = createDispatcher({ + bodyTimeout: requestTimeoutMs, + headersTimeout: requestTimeoutMs, + }); + + return (input, init) => { + if (input instanceof Request) { + const body = init?.body ?? input.body; + + return fetchImpl(input.url, { + ...init, + body, + dispatcher, + ...(body ? { duplex: "half" } : {}), + headers: init?.headers ?? input.headers, + method: init?.method ?? input.method, + signal: init?.signal ?? input.signal, + } as DispatcherRequestInit); + } + + return fetchImpl(input, { + ...init, + dispatcher, + } as DispatcherRequestInit); + }; +} + function resolveUnstructuredApiUrl(env: ApiParserEnv): string | undefined { const configured = env.UNSTRUCTURED_API_URL?.trim(); if (configured) { diff --git a/knowledge-fs/docs/production-deployment.md b/knowledge-fs/docs/production-deployment.md index 349c336b7a0..c8bee4359ce 100644 --- a/knowledge-fs/docs/production-deployment.md +++ b/knowledge-fs/docs/production-deployment.md @@ -64,7 +64,7 @@ the service: | `UNSTRUCTURED_API_URL` | Parser endpoint for complex formats. | | `UNSTRUCTURED_API_KEY` | Optional parser authentication. | | `UNSTRUCTURED_MAX_CONCURRENCY` | Process-wide parser request limit; defaults to `2`. | -| `UNSTRUCTURED_REQUEST_TIMEOUT_MS` | Total timeout for one parser request and response body; defaults to `120000`. | +| `UNSTRUCTURED_REQUEST_TIMEOUT_MS` | Total timeout for one parser request and response body; defaults to `120000` and accepts up to `1800000` for long OCR documents. | | `UNSTRUCTURED_MAX_RESPONSE_BYTES` | Maximum parser response body; defaults to `33554432` (32 MiB). | Compose injects `DIFY_INNER_API_URL` and `DIFY_INNER_API_KEY`; do not duplicate them in the diff --git a/knowledge-fs/packages/api/src/document-compilation-worker.test.ts b/knowledge-fs/packages/api/src/document-compilation-worker.test.ts index 6f054645589..0906a9be2d5 100644 --- a/knowledge-fs/packages/api/src/document-compilation-worker.test.ts +++ b/knowledge-fs/packages/api/src/document-compilation-worker.test.ts @@ -375,7 +375,7 @@ describe("createDocumentCompilationWorker lease integration", () => { let smokeCalls = 0; const outlines = createInMemoryDocumentOutlineRepository({ maxOutlines: 2 }); const knowledgePaths = createInMemoryKnowledgePathRepository({ - maxBatchSize: 10, + maxBatchSize: 2, maxListLimit: 10, maxPaths: 10, }); diff --git a/knowledge-fs/packages/api/src/document-compilation-worker.ts b/knowledge-fs/packages/api/src/document-compilation-worker.ts index e40a25d60a1..d2b804368da 100644 --- a/knowledge-fs/packages/api/src/document-compilation-worker.ts +++ b/knowledge-fs/packages/api/src/document-compilation-worker.ts @@ -618,7 +618,8 @@ export function createDocumentCompilationWorker({ documentOutlineIds = [persistedOutline.id]; if (knowledgePaths && generateKnowledgePathId) { await assertWritable(); - const persistedPaths = await knowledgePaths.upsertMany( + const persistedPaths = await upsertKnowledgePathsInBatches( + knowledgePaths, buildCompilationKnowledgePaths({ asset: activeAsset, generateId: generateKnowledgePathId, @@ -729,7 +730,8 @@ export function createDocumentCompilationWorker({ documentOutlineIds = [persistedOutline.id]; if (knowledgePaths && generateKnowledgePathId) { await assertWritable(); - const persistedPaths = await knowledgePaths.upsertMany( + const persistedPaths = await upsertKnowledgePathsInBatches( + knowledgePaths, buildCompilationKnowledgePaths({ asset: activeAsset, generateId: generateKnowledgePathId, @@ -1353,6 +1355,21 @@ function buildCompilationKnowledgePaths({ ]; } +async function upsertKnowledgePathsInBatches( + repository: KnowledgePathRepository, + paths: readonly KnowledgePath[], +): Promise { + const persisted: KnowledgePath[] = []; + + for (let offset = 0; offset < paths.length; offset += repository.maxBatchSize) { + persisted.push( + ...(await repository.upsertMany(paths.slice(offset, offset + repository.maxBatchSize))), + ); + } + + return persisted; +} + function componentReferences( componentKeys: readonly string[], generationId: string, diff --git a/knowledge-fs/packages/api/src/graph-index-repository.ts b/knowledge-fs/packages/api/src/graph-index-repository.ts index 13f23912d8d..24ca8442fa1 100644 --- a/knowledge-fs/packages/api/src/graph-index-repository.ts +++ b/knowledge-fs/packages/api/src/graph-index-repository.ts @@ -56,6 +56,7 @@ export interface GraphRelation { } export interface GraphIndexRepository { + readonly maxBatchSize: number; deleteComponentsBySourceNodesAcrossGenerations( input: DeleteGraphComponentsBySourceNodesAcrossGenerationsInput, ): Promise; @@ -179,6 +180,7 @@ export function createInMemoryGraphIndexRepository({ const relations = new Map(); return { + maxBatchSize, deleteComponentsBySourceNodesAcrossGenerations: async (input) => { validateGraphPruneSourceNodesAcrossGenerationsInput(input); return deleteInMemoryGraphComponentsBySourceNodes({ @@ -519,6 +521,7 @@ export function createDatabaseGraphIndexRepository({ }); return { + maxBatchSize, deleteComponentsBySourceNodesAcrossGenerations: async (input) => database.transaction((transaction) => deleteDatabaseGraphComponentsBySourceNodesAcrossGenerations(database, transaction, input), diff --git a/knowledge-fs/packages/api/src/graph-index-writer.ts b/knowledge-fs/packages/api/src/graph-index-writer.ts index 6e87bac050d..a6b55737d6f 100644 --- a/knowledge-fs/packages/api/src/graph-index-writer.ts +++ b/knowledge-fs/packages/api/src/graph-index-writer.ts @@ -179,7 +179,10 @@ export function createGraphIndexWriter({ updatedAt: timestamp, }), ); - const storedEntities = await graph.upsertEntities(entityInputs); + const graphBatchSize = Math.min(maxBatchSize, graph.maxBatchSize); + const storedEntities = await persistGraphBatches(entityInputs, graphBatchSize, (batch) => + graph.upsertEntities(batch), + ); // Back-reference: record on each source node the graph entity ids it now maps to, so // retrieval can seed graph expansion from a node's matched entities. `updateMetadataMany` // replaces metadata, so merge onto the node's current metadata. This runs as the last @@ -294,7 +297,9 @@ export function createGraphIndexWriter({ updatedAt: timestamp, }), ); - const storedRelations = await graph.upsertRelations(relationInputs); + const storedRelations = await persistGraphBatches(relationInputs, graphBatchSize, (batch) => + graph.upsertRelations(batch), + ); return { entities: storedEntities.map(cloneGraphEntity), @@ -364,6 +369,20 @@ export function createGraphIndexWriter({ return writer; } +async function persistGraphBatches( + input: readonly T[], + maxBatchSize: number, + persist: (batch: readonly T[]) => Promise, +): Promise { + const persisted: T[] = []; + + for (let offset = 0; offset < input.length; offset += maxBatchSize) { + persisted.push(...(await persist(input.slice(offset, offset + maxBatchSize)))); + } + + return persisted; +} + type GraphQualityFlag = { readonly graphEligible: boolean; readonly reason?: "budget" | "confidence-threshold" | "duplicate" | undefined; diff --git a/knowledge-fs/packages/api/src/graph-index.test.ts b/knowledge-fs/packages/api/src/graph-index.test.ts index e6eb963cfc3..2cc40757c99 100644 --- a/knowledge-fs/packages/api/src/graph-index.test.ts +++ b/knowledge-fs/packages/api/src/graph-index.test.ts @@ -315,14 +315,14 @@ function createFakeGraphTraversalExecutor() { describe("graph index persistence", () => { const graphTimestamp = "2026-05-12T12:00:00.000Z"; - it("indexes graph-eligible entities and relations without writing ineligible outputs", async () => { + it("indexes eligible graph outputs in repository-sized batches", async () => { const nodes = createInMemoryKnowledgeNodeRepository({ maxBatchSize: 4, maxListLimit: 4, maxNodes: 4, }); const graph = createInMemoryGraphIndexRepository({ - maxBatchSize: 8, + maxBatchSize: 1, maxEntities: 8, maxRelations: 8, now: () => "2026-05-12T12:00:00.000Z", @@ -360,6 +360,13 @@ describe("graph index persistence", () => { subject: "Acme Corp", type: "mentions", }, + { + confidence: 0.9, + object: "Refund Policy", + quality: { graphEligible: true }, + subject: "Acme Corp", + type: "references", + }, { confidence: 0.42, object: "weak concept", @@ -405,7 +412,7 @@ describe("graph index persistence", () => { expect(result.missingNodeIds).toEqual(["018f0d60-7a49-7cc2-9c1b-5b36f18f2c99"]); expect(result.stats).toEqual({ entitiesIndexed: 2, - relationsIndexed: 1, + relationsIndexed: 2, skippedEntities: 1, skippedRelations: 1, }); diff --git a/knowledge-fs/packages/api/src/knowledge-path-repository.ts b/knowledge-fs/packages/api/src/knowledge-path-repository.ts index 233bb7bc38d..ff5c615492c 100644 --- a/knowledge-fs/packages/api/src/knowledge-path-repository.ts +++ b/knowledge-fs/packages/api/src/knowledge-path-repository.ts @@ -73,6 +73,7 @@ export interface DeleteKnowledgePathsByDocumentAssetInput { } export interface KnowledgePathRepository { + readonly maxBatchSize: number; create(input: KnowledgePath): Promise; deleteByDocumentAsset(input: DeleteKnowledgePathsByDocumentAssetInput): Promise; deleteSemanticView(input: DeleteSemanticViewPathsInput): Promise; @@ -129,6 +130,7 @@ export function createInMemoryKnowledgePathRepository({ const paths = new Map(); return { + maxBatchSize, create: async (input) => { const path = cloneKnowledgePath(KnowledgePathSchema.parse(input)); const key = knowledgePathKey( @@ -343,6 +345,7 @@ export function createDatabaseKnowledgePathRepository({ const tableName = "knowledge_paths"; return { + maxBatchSize, create: async (input) => { const path = cloneKnowledgePath(KnowledgePathSchema.parse(input)); return path.publicationGenerationId diff --git a/knowledge-fs/packages/api/src/retrieval-paths-coverage.test.ts b/knowledge-fs/packages/api/src/retrieval-paths-coverage.test.ts index 4d0bcb2806f..687953500df 100644 --- a/knowledge-fs/packages/api/src/retrieval-paths-coverage.test.ts +++ b/knowledge-fs/packages/api/src/retrieval-paths-coverage.test.ts @@ -126,6 +126,7 @@ function fakeGraph( traverse: (startEntityId: string) => GraphTraversalResult, ): GraphIndexRepository { return { + maxBatchSize: 1, deleteComponentsBySourceNodesAcrossGenerations: async () => { throw new Error( "deleteComponentsBySourceNodesAcrossGenerations is not used by graph expansion", diff --git a/knowledge-fs/packages/parsers/src/index.ts b/knowledge-fs/packages/parsers/src/index.ts index 5ef1f8e1763..8370f9fd9ec 100644 --- a/knowledge-fs/packages/parsers/src/index.ts +++ b/knowledge-fs/packages/parsers/src/index.ts @@ -193,6 +193,7 @@ const defaultMaxResponseBytes = 32 * 1024 * 1024; const defaultMaxConcurrency = 2; const defaultMaxRetries = 0; const defaultRequestTimeoutMs = 120_000; +const maxRequestTimeoutMs = 1_800_000; const defaultMaxRows = 20_000; const defaultRetryDelayMs = 100; const defaultNow = () => new Date().toISOString(); @@ -3600,10 +3601,10 @@ function validateUnstructuredResourceOptions({ if ( !Number.isSafeInteger(requestTimeoutMs) || requestTimeoutMs < 1 || - requestTimeoutMs > 600_000 + requestTimeoutMs > maxRequestTimeoutMs ) { throw new ProviderInputError( - "Unstructured parser requestTimeoutMs must be an integer between 1 and 600000", + `Unstructured parser requestTimeoutMs must be an integer between 1 and ${maxRequestTimeoutMs}`, ); } } diff --git a/knowledge-fs/packages/parsers/src/parser.test.ts b/knowledge-fs/packages/parsers/src/parser.test.ts index 3ce5be73899..f96810823ad 100644 --- a/knowledge-fs/packages/parsers/src/parser.test.ts +++ b/knowledge-fs/packages/parsers/src/parser.test.ts @@ -3128,13 +3128,13 @@ describe("parser adapters", () => { "maxConcurrency must be an integer between 1 and 32", ); expect(() => createUnstructuredParserClient({ ...base, requestTimeoutMs: 0 })).toThrow( - "requestTimeoutMs must be an integer between 1 and 600000", + "requestTimeoutMs must be an integer between 1 and 1800000", ); - expect(() => createUnstructuredParserClient({ ...base, requestTimeoutMs: 600_001 })).toThrow( - "requestTimeoutMs must be an integer between 1 and 600000", + expect(() => createUnstructuredParserClient({ ...base, requestTimeoutMs: 1_800_001 })).toThrow( + "requestTimeoutMs must be an integer between 1 and 1800000", ); expect(() => createUnstructuredParserClient({ ...base, requestTimeoutMs: 1.5 })).toThrow( - "requestTimeoutMs must be an integer between 1 and 600000", + "requestTimeoutMs must be an integer between 1 and 1800000", ); }); diff --git a/knowledge-fs/pnpm-lock.yaml b/knowledge-fs/pnpm-lock.yaml index a97435140ee..4f329030d44 100644 --- a/knowledge-fs/pnpm-lock.yaml +++ b/knowledge-fs/pnpm-lock.yaml @@ -104,6 +104,9 @@ importers: sharp: specifier: 0.35.3 version: 0.35.3(@types/node@22.19.18) + undici: + specifier: ^8.10.0 + version: 8.10.0 devDependencies: '@types/node': specifier: ^22.10.2 @@ -2207,6 +2210,10 @@ packages: undici-types@6.21.0: resolution: {integrity: sha512-iwDZqg0QAGrg9Rav5H4n0M64c3mkR59cJ6wQp+7C4nI0gsmExaedaYLNO44eT4AtBBwjbTiGPMlt2Md0T9H9JQ==} + undici@8.10.0: + resolution: {integrity: sha512-HvltHd7avK13QIw/oLe4qoOLyoVSoafqJ2jYOrtMRBkbYT31eiBQ8O0ehRKZiEZCMEyLFQNIADpgCWC5fALvYQ==} + engines: {node: '>=22.19.0'} + unicode-segmenter@0.15.0: resolution: {integrity: sha512-Xmvwqx4F8nGuCv2eGPJVJq73NMTfpqx2Xe9/v5hQoyAUnERVhX+sRkyYVdYoBUbnTok2FTBOlstUeQ5sRleXSA==} @@ -3883,6 +3890,8 @@ snapshots: undici-types@6.21.0: {} + undici@8.10.0: {} + unicode-segmenter@0.15.0: {} unpipe@1.0.0: {}