504 Gateway Timeout Error When Upserting Documents into Vector Stor

[AHMEDBASHIR999]

Describe the bug

I am experiencing a 504 Gateway Timeout (Cloudflare) error while attempting to upsert data into the vector store in Flowise AI.

To Reproduce

Steps to Reproduce:
Open Flowise canvas.
Connect:
Text File node
Recursive Character Text Splitter
Upsert Vector Store
Upload a document.
Click to run the upsert process.

Expected behavior

The document should be split, embeddings generated, and successfully stored in the vector database.

Screenshots

Flow

No response

Use Method

None

Flowise Version

No response

Operating System

None

Browser

None

Additional context

No response

[darshjme-codes]

Expert-Level Analysis and Solution for GitHub Issue #5767

1. Root Cause Analysis

The 504 Gateway Timeout error during vector store upsert operations indicates a fundamental architectural bottleneck in Flowise's request handling pipeline. This occurs due to several interconnected issues:

Primary Root Cause: The synchronous processing model where large document uploads trigger blocking operations that exceed Cloudflare's default 100-second timeout threshold. When documents are processed through the Recursive Character Text Splitter → Embedding Generation → Vector Storage pipeline, each step can take considerable time, especially for large documents with complex embedding models.

Secondary Issues:

• Memory Management: Large documents consume significant RAM during text splitting and embedding generation, potentially causing garbage collection pauses
• Database Connection Pooling: Vector databases like Pinecone, Weaviate, or ChromaDB may have connection limits that create bottlenecks during bulk upserts
• Embedding API Rate Limits: External embedding services (OpenAI, Hugging Face) impose rate limits that compound the timeout issue
• Lack of Progress Tracking: No intermediate status updates to keep the connection alive during long-running operations

The architectural flaw lies in treating document upsert as a single atomic HTTP request rather than implementing an asynchronous job queue pattern with proper progress tracking and timeout management.

2. Three Solution Options (Ranked)

Option 1: Quick Fix (Recommended)

Implement request timeout extension and chunked processing with heartbeat responses:

// flowise/packages/server/src/routes/document-upsert.ts
import { Request, Response } from 'express';
import { setTimeout } from 'timers/promises';

export const handleDocumentUpsert = async (req: Request, res: Response) => {
    // Extend socket timeout for this specific route
    req.socket.setTimeout(300000); // 5 minutes
    
    try {
        // Send initial response to prevent gateway timeout
        res.writeHead(200, {
            'Content-Type': 'application/json',
            'Connection': 'keep-alive',
            'Cache-Control': 'no-cache'
        });
        
        res.write(JSON.stringify({ status: 'processing', message: 'Starting document upsert...' }));
        
        // Process in chunks with periodic heartbeats
        const result = await processDocumentInChunks(req.body, (progress) => {
            res.write(JSON.stringify({ 
                status: 'in-progress', 
                progress: progress,
                timestamp: new Date().toISOString()
            }) + '\n');
        });
        
        res.write(JSON.stringify({ 
            status: 'completed', 
            result: result 
        }));
        res.end();
        
    } catch (error) {
        res.write(JSON.stringify({ 
            status: 'error', 
            error: error.message 
        }));
        res.end();
    }
};

async function processDocumentInChunks(data: any, onProgress: (progress: number) => void) {
    const totalSteps = 5; // text splitting, embedding, upsert preparation, upsert execution, completion
    let currentStep = 0;
    
    // Simulate chunked processing with progress updates
    for (let i = 0; i < totalSteps; i++) {
        // Actual processing logic here
        await new Promise(resolve => setTimeout(resolve, 1000)); // Simulate work
        currentStep++;
        onProgress((currentStep / totalSteps) * 100);
    }
    
    return { success: true, documentsProcessed: data.documents.length };
}

Option 2: Proper Fix (Architectural)

Implement background job processing with Redis queue and WebSocket progress tracking:

// flowise/packages/server/src/services/document-processing.service.ts
import { Queue, Worker, Job } from 'bullmq';
import Redis from 'ioredis';
import { v4 as uuidv4 } from 'uuid';

interface DocumentUpsertJobData {
    documentId: string;
    userId: string;
    fileBuffer: Buffer;
    embeddingConfig: any;
    vectorStoreConfig: any;
}

interface DocumentUpsertJobResult {
    success: boolean;
    documentCount: number;
    embeddingTimeMs: number;
    upsertTimeMs: number;
    error?: string;
}

export class DocumentProcessingService {
    private queue: Queue<DocumentUpsertJobData, DocumentUpsertJobResult>;
    private worker: Worker<DocumentUpsertJobData, DocumentUpsertJobResult>;
    private redis: Redis;

    constructor() {
        this.redis = new Redis(process.env.REDIS_URL || 'redis://localhost:6379');
        this.queue = new Queue<DocumentUpsertJobData, DocumentUpsertJobResult>('document-upsert', {
            connection: this.redis,
            defaultJobOptions: {
                attempts: 3,
                backoff: { type: 'exponential', delay: 1000 },
                timeout: 300000, // 5 minutes
            }
        });

        this.worker = new Worker<DocumentUpsertJobData, DocumentUpsertJobResult>(
            'document-upsert',
            this.processDocument.bind(this),
            { connection: this.redis }
        );
    }

    async submitDocumentUpsert(data: DocumentUpsertJobData): Promise<string> {
        const jobId = uuidv4();
        await this.queue.add('upsert-document', data, { jobId });
        return jobId;
    }

    private async processDocument(job: Job<DocumentUpsertJobData>): Promise<DocumentUpsertJobResult> {
        const startTime = Date.now();
        
        try {
            // Update job progress
            await job.updateProgress(10);
            
            // Step 1: Text splitting
            const splitDocuments = await this.splitText(job.data.fileBuffer);
            await job.updateProgress(30);
            
            // Step 2: Generate embeddings
            const embeddingStartTime = Date.now();
            const embeddedDocuments = await this.generateEmbeddings(splitDocuments, job.data.embeddingConfig);
            const embeddingTime = Date.now() - embeddingStartTime;
            await job.updateProgress(70);
            
            // Step 3: Upsert to vector store
            const upsertStartTime = Date.now();
            await this.upsertToVectorStore(embeddedDocuments, job.data.vectorStoreConfig);
            const upsertTime = Date.now() - upsertStartTime;
            
            return {
                success: true,
                documentCount: embeddedDocuments.length,
                embeddingTimeMs: embeddingTime,
                upsertTimeMs: upsertTime
            };
        } catch (error) {
            return {
                success: false,
                documentCount: 0,
                embeddingTimeMs: 0,
                upsertTimeMs: 0,
                error: error.message
            };
        }
    }

    private async splitText(buffer: Buffer): Promise<any[]> {
        // Implementation of text splitting logic
        return [];
    }

    private async generateEmbeddings(documents: any[], config: any): Promise<any[]> {
        // Implementation of embedding generation
        return [];
    }

    private async upsertToVectorStore(documents: any[], config: any): Promise<void> {
        // Implementation of vector store upsert
    }
}

Option 3: Long-Term (Best Practice)

Complete rearchitecture with streaming, microservices, and comprehensive monitoring:

// flowise/packages/core/src/pipeline/streaming-pipeline.ts
import { Transform, Readable, Writable } from 'stream';
import { EventEmitter } from 'events';
import { MetricsCollector } from './metrics-collector';

export interface PipelineConfig {
    maxConcurrency: number;
    timeoutMs: number;
    retryAttempts: number;
    batchSize: number;
    memoryLimit: number;
}

export class StreamingDocumentPipeline extends EventEmitter {
    private config: PipelineConfig;
    private metrics: MetricsCollector;

    constructor(config: PipelineConfig) {
        super();
        this.config = config;
        this.metrics = new MetricsCollector();
    }

    async processDocumentStream(inputStream: Readable, outputStream: Writable): Promise<void> {
        const controller = new AbortController();
        
        // Create processing pipeline
        const textSplitter = new TextSplitTransform({
            chunkSize: 1000,
            overlap: 200
        });

        const embeddingGenerator = new EmbeddingTransform({
            model: 'text-embedding-ada-002',
            batchSize: this.config.batchSize
        });

        const vectorUpsert = new VectorUpsertTransform({
            storeType: 'chromadb',
            batchTimeout: 5000
        });

        // Wire up the pipeline with error handling and monitoring
        inputStream
            .pipe(textSplitter)
            .on('error', (err) => this.handleError(err))
            .pipe(embeddingGenerator)
            .on('error', (err) => this.handleError(err))
            .pipe(vectorUpsert)
            .on('error', (err) => this.handleError(err))
            .pipe(outputStream);

        // Monitor progress and emit events
        this.monitorPipeline(inputStream, textSplitter, embeddingGenerator, vectorUpsert);
    }

    private monitorPipeline(...transforms: Transform[]) {
        transforms.forEach((transform, index) => {
            transform.on('data', () => {
                this.metrics.increment(`pipeline.stage.${index}.processed`);
            });
        });
    }

    private handleError(error: Error) {
        this.emit('error', error);
        this.metrics.increment('pipeline.errors.total');
    }
}

class TextSplitTransform extends Transform {
    private chunkSize: number;
    private overlap: number;
    private buffer: string = '';

    constructor(options: { chunkSize: number; overlap: number }) {
        super({ objectMode: true });
        this.chunkSize = options.chunkSize;
        this.overlap = options.overlap;
    }

    _transform(chunk: Buffer, encoding: string, callback: Function) {
        this.buffer += chunk.toString();
        
        // Process buffer into chunks
        const chunks = this.createChunks(this.buffer);
        this.buffer = chunks.remaining;
        
        chunks.processed.forEach(textChunk => {
            callback(null, { text: textChunk, metadata: {} });
        });
    }

    private createChunks(text: string): { processed: string[]; remaining: string } {
        const chunks: string[] = [];
        let position = 0;

        while (position < text.length) {
            const end = Math.min(position + this.chunkSize, text.length);
            const chunk = text.substring(position, end);
            chunks.push(chunk);
            position = end - this.overlap;
        }

        return {
            processed: chunks.slice(0, -1), // All but last (incomplete)
            remaining: chunks[chunks.length - 1] || ''
        };
    }
}

3. Production Test Code

// tests/integration/document-upsert-timeout.test.ts
import { describe, it, beforeEach, afterEach } from 'vitest';
import { DocumentProcessingService } from '../../src/services/document-processing.service';
import { createMockServer } from '../utils/mock-server';
import { generateLargeTestDocument } from '../utils/test-data';
import { promisify } from 'util';
import { setTimeout } from 'timers/promises';

describe('Document Upsert Timeout Handling', () => {
    let service: DocumentProcessingService;
    let mockServer: any;

    beforeEach(async () => {
        mockServer = await createMockServer();
        service = new DocumentProcessingService();
    });

    afterEach(async () => {
        await mockServer.close();
        await service.shutdown();
    });

    it('should handle large document upsert without timeout', async () => {
        const largeDocument = generateLargeTestDocument(10000); // 10k words
        
        const jobId = await service.submitDocumentUpsert({
            documentId: 'test-doc-1',
            userId: 'test-user',
            fileBuffer: Buffer.from(largeDocument),
            embeddingConfig: { model: 'test-model' },
            vectorStoreConfig: { type: 'test-store' }
        });

        // Wait for job completion with timeout
        const result = await waitForJobCompletion(service, jobId, 300000); // 5 minutes
        
        expect(result.success).toBe(true);
        expect(result.documentCount).toBeGreaterThan(0);
    });

    it('should provide progress updates during processing', async () => {
        const progressUpdates: number[] = [];
        const largeDocument = generateLargeTestDocument(5000);

        // Mock progress tracking
        const originalUpdateProgress = jest.spyOn(Job.prototype, 'updateProgress');
        originalUpdateProgress.mockImplementation(async (progress: number) => {
            progressUpdates.push(progress);
        });

        const jobId = await service.submitDocumentUpsert({
            documentId: 'test-doc-2',
            userId: 'test-user',
            fileBuffer: Buffer.from(largeDocument),
            embeddingConfig: { model: 'test-model' },
            vectorStoreConfig: { type: 'test-store' }
        });

        await waitForJobCompletion(service, jobId, 300000);
        
        // Verify progress was updated multiple times
        expect(progressUpdates.length).toBeGreaterThan(3);
        expect(Math.max(...progressUpdates)).toBe(100);
    });

    it('should handle embedding API failures gracefully', async () => {
        // Mock embedding service failure
        jest.spyOn(DocumentProcessingService.prototype as any, 'generateEmbeddings')
            .mockRejectedValue(new Error('Embedding API timeout'));

        const jobId = await service.submitDocumentUpsert({
            documentId: 'test-doc-3',
            userId: 'test-user',
            fileBuffer: Buffer.from('test content'),
            embeddingConfig: { model: 'failing-model' },
            vectorStoreConfig: { type: 'test-store' }
        });

        const result = await waitForJobCompletion(service, jobId, 60000);
        
        expect(result.success).toBe(false);
        expect(result.error).toContain('Embedding API');
    });

    it('should respect memory limits during processing', async () => {
        const largeDocument = generateLargeTestDocument(50000); // Very large document
        
        // Test with memory monitoring
        const memoryBefore = process.memoryUsage().heapUsed;
        const jobId = await service.submitDocumentUpsert({
            documentId: 'test-doc-4',
            userId: 'test-user',
            fileBuffer: Buffer.from(largeDocument),
            embeddingConfig: { model: 'test-model' },
            vectorStoreConfig: { type: 'test-store' }
        });

        const result = await waitForJobCompletion(service, jobId, 600000); // 10 minutes
        const memoryAfter = process.memoryUsage().heapUsed;
        
        // Memory usage should not exceed reasonable bounds
        const memoryGrowth = memoryAfter - memoryBefore;
        expect(memoryGrowth).toBeLessThan(500 * 1024 * 1024); // Less than 500MB
    });

    it('should handle concurrent document upserts', async () => {
        const promises = [];
        for (let i = 0; i < 5; i++) {
            const doc = generateLargeTestDocument(2000);
            promises.push(service.submitDocumentUpsert({
                documentId: `test-doc-${i}`,
                userId: 'test-user',
                fileBuffer: Buffer.from(doc),
                embeddingConfig: { model: 'test-model' },
                vectorStoreConfig: { type: 'test-store' }
            }));
        }

        const jobIds = await Promise.all(promises);
        
        // Wait for all jobs to complete
        const results = await Promise.all(
            jobIds.map(jobId => waitForJobCompletion(service, jobId, 120000))
        );

        // All should succeed
        results.forEach(result => {
            expect(result.success).toBe(true);
        });
    });
});

async function waitForJobCompletion(service: DocumentProcessingService, jobId: string, timeoutMs: number) {
    const startTime = Date.now();
    
    while (Date.now() - startTime < timeoutMs) {
        const job = await service.getJob(jobId);
        if (job && job.isCompleted()) {
            return job.returnvalue;
        }
        await setTimeout(1000); // Wait 1 second before checking again
    }
    
    throw new Error(`Job ${jobId} did not complete within ${timeoutMs}ms`);
}

4. Impact Assessment

• Severity: HIGH - Affects core functionality for document processing workflows
• Affected Users: Users working with medium to large documents (>10 pages) or using slower embedding models
• Consequences: Complete workflow failure, user frustration, potential data loss during interrupted upserts
• Migration Path: Option 1 requires no breaking changes. Options 2-3 introduce async job patterns that may require UI updates but maintain API compatibility

5. Recommended Implementation Strategy

1. Immediate: Deploy Option 1 (Quick Fix) to resolve immediate timeout issues
2. Short-term: Implement Option 2 (Background Jobs) for production environments
3. Long-term: Plan migration to Option 3 (Streaming Architecture) for enterprise scalability

---

Built by @darshjme-codes — Creator of Samast (universal LLM provider abstraction) and Brahmand CLI.

[autonomy414941]

This looks like an edge timeout rather than a parser bug.

504 Gateway Timeout (Cloudflare) usually means Cloudflare timed out waiting for origin response (commonly around ~100s), and vector upsert can easily run longer than that.

I checked Flowise code path for the canvas upsert button:

• UI calls POST /api/v1/vector/internal-upsert/:id
• server waits for upsert completion before responding (job.waitUntilFinished(...) in queue mode too)

So if upsert takes > Cloudflare timeout, client gets 504 even if Flowise eventually finishes.

Quick way to confirm:

# 1) Run upsert through origin directly (bypass Cloudflare)
# from the host/network where Flowise is reachable
curl -i -X POST "http://<flowise-origin>:3000/api/v1/vector/internal-upsert/<chatflowId>" \
  -H "Authorization: Bearer <token>" \
  -H "Content-Type: application/json" \
  --data '{"stopNodeId":"<vector-node-id>"}'

# 2) In parallel, watch logs
docker logs -f <flowise-container>

If direct-origin request succeeds while Cloudflare route 504s, mitigation is to bypass Cloudflare proxy for the Flowise API host/path used for upsert (DNS-only origin host or internal/VPN URL for admin upserts).

Also please share:

• Flowise version
• vector DB type
• embedding model/provider
• approx document size/chunk settings

With those, we can suggest concrete batching/chunk tuning next.