diff --git a/Dockerfile b/Dockerfile index 25ade7ea2ca..5d3be512a63 100644 --- a/Dockerfile +++ b/Dockerfile @@ -29,7 +29,7 @@ RUN pnpm config set store-dir /root/.pnpm-store ENV PUPPETEER_SKIP_DOWNLOAD=true ENV PUPPETEER_EXECUTABLE_PATH=/usr/bin/chromium-browser -ENV NODE_OPTIONS=--max-old-space-size=8192 +ENV NODE_OPTIONS=--max-old-space-size=4096 ################################################################################ # Prune projects diff --git a/packages/components/nodes/documentloaders/AAIDomains/AAIDomains.ts b/packages/components/nodes/documentloaders/AAIDomains/AAIDomains.ts index 3df934eb6cc..8f04e358c39 100644 --- a/packages/components/nodes/documentloaders/AAIDomains/AAIDomains.ts +++ b/packages/components/nodes/documentloaders/AAIDomains/AAIDomains.ts @@ -261,16 +261,9 @@ class AAIDomains_DocumentLoaders implements INode { contentFields: parsedContentFields } - const loader = new AAIDomainsLoader(loaderOptions) + const loader = new AAIDomainsLoader({ ...loaderOptions, textSplitter: textSplitter || null }) - let docs: IDocument[] = [] - - if (textSplitter) { - docs = await loader.load() - docs = await textSplitter.splitDocuments(docs) - } else { - docs = await loader.load() - } + const docs = await loader.load() // Apply metadata const parsedMetadata = metadata ? (typeof metadata === 'object' ? metadata : JSON.parse(metadata)) : null @@ -305,6 +298,7 @@ interface AAIDomainsLoaderParams { isValid: string | null hasAnalysis: string contentFields: string[] | null + textSplitter?: TextSplitter | null } class AAIDomainsLoader extends BaseDocumentLoader { @@ -318,6 +312,7 @@ class AAIDomainsLoader extends BaseDocumentLoader { private isValid: string | null private hasAnalysis: string private contentFields: string[] | null + private textSplitter: TextSplitter | null constructor(params: AAIDomainsLoaderParams) { super() @@ -331,6 +326,7 @@ class AAIDomainsLoader extends BaseDocumentLoader { this.isValid = params.isValid this.hasAnalysis = params.hasAnalysis this.contentFields = params.contentFields + this.textSplitter = params.textSplitter ?? null } public async load(): Promise { @@ -350,10 +346,11 @@ class AAIDomainsLoader extends BaseDocumentLoader { } }) - // Use larger page size since we're only selecting essential fields const pageSize = Math.min(this.limit, 100) let allDocs: IDocument[] = [] - let currentPage = 0 + let fetchedDomainCount = 0 + let lastId: string | null = null + let pageNum = 0 console.info('[AAIDomains] Starting load with params:', { limit: this.limit, @@ -365,11 +362,11 @@ class AAIDomainsLoader extends BaseDocumentLoader { hasAnalysis: this.hasAnalysis }) - while (allDocs.length < this.limit) { - const remainingItems = this.limit - allDocs.length + while (fetchedDomainCount < this.limit) { + const remainingItems = this.limit - fetchedDomainCount const currentPageSize = Math.min(pageSize, remainingItems) - console.info(`[AAIDomains] Fetching page ${currentPage}, size ${currentPageSize}`) + console.info(`[AAIDomains] Fetching page ${pageNum}, size ${currentPageSize}`) // Retry logic with exponential backoff let retryCount = 0 @@ -446,8 +443,12 @@ class AAIDomainsLoader extends BaseDocumentLoader { let query = supabase .from('domains') .select(selectFields) - .order('updated_at', { ascending: false }) - .range(currentPage * pageSize, (currentPage + 1) * pageSize - 1) + .order('id', { ascending: true }) + .limit(currentPageSize) + + if (lastId) { + query = query.gt('id', lastId) + } // Apply filters if (this.searchTerm) { @@ -479,7 +480,7 @@ class AAIDomainsLoader extends BaseDocumentLoader { throw new Error(`Failed to fetch domains from AAI Datastore: ${error.message}`) } - console.info(`[AAIDomains] Page ${currentPage} response:`, { + console.info(`[AAIDomains] Page ${pageNum} response:`, { hasData: !!data, domainsCount: data?.length || 0 }) @@ -502,17 +503,26 @@ class AAIDomainsLoader extends BaseDocumentLoader { filteredDomains = this.filterByTags(domainsWithTags) } + fetchedDomainCount += filteredDomains.length + const pageDocs = filteredDomains.map((d: any) => this.createDocumentFromDomain(d)) - allDocs.push(...pageDocs) - currentPage++ - // Stop if we've fetched enough - if (allDocs.length >= this.limit || data.length < currentPageSize) { + if (this.textSplitter) { + const pageChunks = await this.textSplitter.splitDocuments(pageDocs) + allDocs.push(...pageChunks) + } else { + allDocs.push(...pageDocs) + } + + // Advance cursor only after docs are successfully pushed + lastId = (data as any[])[data.length - 1].id + pageNum++ + + if (fetchedDomainCount >= this.limit || data.length < currentPageSize) { shouldStopPagination = true break } - // Success - break out of retry loop break } catch (error: any) { lastError = error @@ -541,12 +551,7 @@ class AAIDomainsLoader extends BaseDocumentLoader { await new Promise((resolve) => setTimeout(resolve, 200)) } - console.info(`[AAIDomains] Load complete. Total documents: ${allDocs.length}`) - - // Truncate to exact limit - if (allDocs.length > this.limit) { - allDocs = allDocs.slice(0, this.limit) - } + console.info(`[AAIDomains] Load complete. Total documents: ${allDocs.length}, source domains fetched: ${fetchedDomainCount}`) return allDocs } diff --git a/packages/server/src/services/documentstore/index.ts b/packages/server/src/services/documentstore/index.ts index 24dcfc9e095..98999aba6ae 100644 --- a/packages/server/src/services/documentstore/index.ts +++ b/packages/server/src/services/documentstore/index.ts @@ -58,6 +58,9 @@ import { Telemetry } from '../../utils/telemetry' import nodesService from '../nodes' // Batch sizes for chunk DB operations and vector store upsert +// PostgreSQL parameter limit is 65535. DocumentStoreFileChunk has 8 columns, so +// SAVE_BATCH_SIZE=500 uses 4000 parameters (6% of limit) — safe headroom. +// If the entity gains columns or this value is raised, recalculate: rows × columns < 65535. const SAVE_BATCH_SIZE = 500 const UPSERT_BATCH_SIZE = 500 const DELETE_BATCH_SIZE = 1000 @@ -465,23 +468,17 @@ const syncAndRefreshChunks = async (storeId: string, fileId: string, userId: str for (let i = 0; i < docs.length; i += SAVE_BATCH_SIZE) { const batch = docs.slice(i, i + SAVE_BATCH_SIZE) try { - await Promise.all( - batch.map(async (chunk: IDocument, localIndex: number) => { - const globalIndex = i + localIndex - const docChunk: DocumentStoreFileChunk = { - userId, - organizationId, - docId: fileId, - storeId: storeId, - id: uuidv4(), - chunkNo: globalIndex + 1, - pageContent: chunk.pageContent, - metadata: JSON.stringify(chunk.metadata) - } - const dChunk = chunkRepository.create(docChunk) - await chunkRepository.save(dChunk) - }) - ) + const entities = batch.map((chunk: IDocument, localIndex: number) => ({ + userId, + organizationId, + docId: fileId, + storeId: storeId, + id: uuidv4(), + chunkNo: i + localIndex + 1, + pageContent: sanitizeChunkContent(chunk.pageContent), + metadata: JSON.stringify(chunk.metadata) + })) + await chunkRepository.insert(entities) // insert() skips TypeORM lifecycle hooks — safe: DocumentStoreFileChunk has none persistedChunks += batch.length persistedChars += batch.reduce((acc: number, chunk: IDocument) => acc + (chunk.pageContent?.length ?? 0), 0) // Free memory: allow GC to reclaim saved chunks @@ -1303,26 +1300,19 @@ const _saveChunksToStorage = async ( for (let i = 0; i < docs.length; i += SAVE_BATCH_SIZE) { const batch = docs.slice(i, i + SAVE_BATCH_SIZE) try { - await Promise.all( - batch.map(async (chunk: IDocument, localIndex: number) => { - const globalIndex = i + localIndex - const docChunk: DocumentStoreFileChunk = { - docId: newLoaderId, - storeId: data.storeId || '', - id: uuidv4(), - chunkNo: globalIndex + 1, - pageContent: sanitizeChunkContent(chunk.pageContent), - metadata: JSON.stringify(chunk.metadata), - userId: data.userId, - organizationId: data.organizationId - } - const dChunk = chunkRepository.create(docChunk) - await chunkRepository.save(dChunk) - }) - ) + const entities = batch.map((chunk: IDocument, localIndex: number) => ({ + docId: newLoaderId, + storeId: data.storeId || '', + id: uuidv4(), + chunkNo: i + localIndex + 1, + pageContent: sanitizeChunkContent(chunk.pageContent), + metadata: JSON.stringify(chunk.metadata), + userId: data.userId, + organizationId: data.organizationId + })) + await chunkRepository.insert(entities) // insert() skips TypeORM lifecycle hooks — safe: DocumentStoreFileChunk has none persistedChunks += batch.length persistedChars += batch.reduce((acc: number, chunk: IDocument) => acc + (chunk.pageContent?.length ?? 0), 0) - // Free memory: allow GC to reclaim saved chunks for (let j = i; j < i + batch.length; j++) { docs[j] = null as any }