From 19bc08cda0bf80439f7cbe261306050b10e34036 Mon Sep 17 00:00:00 2001 From: hannesrudolph Date: Fri, 4 Jul 2025 15:38:33 -0600 Subject: [PATCH] fix: resolve Qdrant connection errors during indexing (#5356) - Add request timeout configuration (30s default) to QdrantClient - Implement retry logic with exponential backoff for socket errors - Add payload size-based chunking to prevent oversized requests (10MB limit) - Implement adaptive batch sizing in FileWatcher based on payload size - Add configuration options for Qdrant timeout settings - Update tests to handle new timeout parameter This fixes the 'SocketError: other side closed' issue that occurs when uploading large payloads (~150MB) to Qdrant during indexing. --- src/services/code-index/config-manager.ts | 15 +- src/services/code-index/constants/index.ts | 4 + .../code-index/processors/file-watcher.ts | 9 +- src/services/code-index/service-factory.ts | 11 +- .../__tests__/qdrant-client.spec.ts | 175 ++++++++++++++++++ .../code-index/vector-store/qdrant-client.ts | 136 +++++++++++--- 6 files changed, 315 insertions(+), 35 deletions(-) diff --git a/src/services/code-index/config-manager.ts b/src/services/code-index/config-manager.ts index d665d2330c..2890abd836 100644 --- a/src/services/code-index/config-manager.ts +++ b/src/services/code-index/config-manager.ts @@ -20,6 +20,7 @@ export class CodeIndexConfigManager { private qdrantUrl?: string = "http://localhost:6333" private qdrantApiKey?: string private searchMinScore?: number + private _qdrantTimeout?: number constructor(private readonly contextProxy: ContextProxy) { // Initialize with current configuration to avoid false restart triggers @@ -358,13 +359,25 @@ export class CodeIndexConfigManager { return this.embedderProvider } + /** + * Gets the Qdrant timeout configuration + */ + public get qdrantTimeout(): number { + if (this._qdrantTimeout === undefined) { + // Use a hardcoded default for now since we don't have a UI setting for this + this._qdrantTimeout = 30000 // 30 seconds default + } + return this._qdrantTimeout + } + /** * Gets the current Qdrant configuration */ - public get qdrantConfig(): { url?: string; apiKey?: string } { + public get qdrantConfig(): { url?: string; apiKey?: string; timeout?: number } { return { url: this.qdrantUrl, apiKey: this.qdrantApiKey, + timeout: this.qdrantTimeout, } } diff --git a/src/services/code-index/constants/index.ts b/src/services/code-index/constants/index.ts index fc66571fbb..a652f93efd 100644 --- a/src/services/code-index/constants/index.ts +++ b/src/services/code-index/constants/index.ts @@ -26,3 +26,7 @@ export const BATCH_PROCESSING_CONCURRENCY = 10 /**Gemini Embedder */ export const GEMINI_MAX_ITEM_TOKENS = 2048 + +/**Qdrant */ +export const MAX_PAYLOAD_SIZE_BYTES = 10 * 1024 * 1024 // 10MB per request +export const QDRANT_REQUEST_TIMEOUT_MS = 30000 // 30 seconds diff --git a/src/services/code-index/processors/file-watcher.ts b/src/services/code-index/processors/file-watcher.ts index 9a1fc3c9af..bb468ed2b7 100644 --- a/src/services/code-index/processors/file-watcher.ts +++ b/src/services/code-index/processors/file-watcher.ts @@ -319,8 +319,13 @@ export class FileWatcher implements IFileWatcher { ): Promise { if (pointsForBatchUpsert.length > 0 && this.vectorStore && !overallBatchError) { try { - for (let i = 0; i < pointsForBatchUpsert.length; i += BATCH_SEGMENT_THRESHOLD) { - const batch = pointsForBatchUpsert.slice(i, i + BATCH_SEGMENT_THRESHOLD) + // Use adaptive batching based on payload size + const estimatedSize = JSON.stringify(pointsForBatchUpsert).length + const pointsPerMB = Math.max(1, Math.floor(pointsForBatchUpsert.length / (estimatedSize / 1024 / 1024))) + const adaptiveBatchSize = Math.min(BATCH_SEGMENT_THRESHOLD, pointsPerMB * 5) // Target 5MB batches + + for (let i = 0; i < pointsForBatchUpsert.length; i += adaptiveBatchSize) { + const batch = pointsForBatchUpsert.slice(i, i + adaptiveBatchSize) let retryCount = 0 let upsertError: Error | undefined diff --git a/src/services/code-index/service-factory.ts b/src/services/code-index/service-factory.ts index 2c00d96525..00a529b243 100644 --- a/src/services/code-index/service-factory.ts +++ b/src/services/code-index/service-factory.ts @@ -108,8 +108,15 @@ export class CodeIndexServiceFactory { throw new Error("Qdrant URL missing for vector store creation") } - // Assuming constructor is updated: new QdrantVectorStore(workspacePath, url, vectorSize, apiKey?) - return new QdrantVectorStore(this.workspacePath, config.qdrantUrl, vectorSize, config.qdrantApiKey) + // Get the full qdrant config including timeout + const qdrantConfig = this.configManager.qdrantConfig + return new QdrantVectorStore( + this.workspacePath, + qdrantConfig.url!, + vectorSize, + qdrantConfig.apiKey, + qdrantConfig.timeout, + ) } /** diff --git a/src/services/code-index/vector-store/__tests__/qdrant-client.spec.ts b/src/services/code-index/vector-store/__tests__/qdrant-client.spec.ts index 798fee4d38..88a0f5366a 100644 --- a/src/services/code-index/vector-store/__tests__/qdrant-client.spec.ts +++ b/src/services/code-index/vector-store/__tests__/qdrant-client.spec.ts @@ -65,6 +65,7 @@ describe("QdrantVectorStore", () => { headers: { "User-Agent": "Roo-Code", }, + timeout: 30000, // Default timeout }) expect(createHash).toHaveBeenCalledWith("sha256") expect(mockCreateHashInstance.update).toHaveBeenCalledWith(mockWorkspacePath) @@ -84,6 +85,7 @@ describe("QdrantVectorStore", () => { headers: { "User-Agent": "Roo-Code", }, + timeout: 30000, // Default timeout }) }) @@ -98,6 +100,29 @@ describe("QdrantVectorStore", () => { headers: { "User-Agent": "Roo-Code", }, + timeout: 30000, // Default timeout + }) + }) + + it("should handle constructor with custom timeout", () => { + const customTimeout = 60000 // 60 seconds + const vectorStoreWithTimeout = new QdrantVectorStore( + mockWorkspacePath, + mockQdrantUrl, + mockVectorSize, + mockApiKey, + customTimeout, + ) + + expect(QdrantClient).toHaveBeenLastCalledWith({ + host: "mock-qdrant", + https: false, + port: 6333, + apiKey: mockApiKey, + headers: { + "User-Agent": "Roo-Code", + }, + timeout: customTimeout, }) }) @@ -118,6 +143,7 @@ describe("QdrantVectorStore", () => { headers: { "User-Agent": "Roo-Code", }, + timeout: 30000, // Default timeout }) expect((vectorStore as any).qdrantUrl).toBe("https://qdrant.ashbyfam.com") }) @@ -133,6 +159,7 @@ describe("QdrantVectorStore", () => { headers: { "User-Agent": "Roo-Code", }, + timeout: 30000, // Default timeout }) expect((vectorStore as any).qdrantUrl).toBe("https://example.com:9000") }) @@ -152,6 +179,7 @@ describe("QdrantVectorStore", () => { headers: { "User-Agent": "Roo-Code", }, + timeout: 30000, // Default timeout }) expect((vectorStore as any).qdrantUrl).toBe("https://example.com/api/v1?key=value") }) @@ -169,6 +197,7 @@ describe("QdrantVectorStore", () => { headers: { "User-Agent": "Roo-Code", }, + timeout: 30000, // Default timeout }) expect((vectorStore as any).qdrantUrl).toBe("http://example.com") }) @@ -184,6 +213,7 @@ describe("QdrantVectorStore", () => { headers: { "User-Agent": "Roo-Code", }, + timeout: 30000, // Default timeout }) expect((vectorStore as any).qdrantUrl).toBe("http://localhost:8080") }) @@ -203,6 +233,7 @@ describe("QdrantVectorStore", () => { headers: { "User-Agent": "Roo-Code", }, + timeout: 30000, // Default timeout }) expect((vectorStore as any).qdrantUrl).toBe("http://example.com/api/v1?key=value") }) @@ -219,6 +250,7 @@ describe("QdrantVectorStore", () => { headers: { "User-Agent": "Roo-Code", }, + timeout: 30000, // Default timeout }) expect((vectorStore as any).qdrantUrl).toBe("http://qdrant.example.com") }) @@ -233,6 +265,7 @@ describe("QdrantVectorStore", () => { headers: { "User-Agent": "Roo-Code", }, + timeout: 30000, // Default timeout }) expect((vectorStore as any).qdrantUrl).toBe("http://localhost:6333") }) @@ -247,6 +280,7 @@ describe("QdrantVectorStore", () => { headers: { "User-Agent": "Roo-Code", }, + timeout: 30000, // Default timeout }) expect((vectorStore as any).qdrantUrl).toBe("http://localhost:9000") }) @@ -263,6 +297,7 @@ describe("QdrantVectorStore", () => { headers: { "User-Agent": "Roo-Code", }, + timeout: 30000, // Default timeout }) expect((vectorStore as any).qdrantUrl).toBe("http://192.168.1.100") }) @@ -277,6 +312,7 @@ describe("QdrantVectorStore", () => { headers: { "User-Agent": "Roo-Code", }, + timeout: 30000, // Default timeout }) expect((vectorStore as any).qdrantUrl).toBe("http://192.168.1.100:6333") }) @@ -293,6 +329,7 @@ describe("QdrantVectorStore", () => { headers: { "User-Agent": "Roo-Code", }, + timeout: 30000, // Default timeout }) expect((vectorStore as any).qdrantUrl).toBe("http://localhost:6333") }) @@ -307,6 +344,7 @@ describe("QdrantVectorStore", () => { headers: { "User-Agent": "Roo-Code", }, + timeout: 30000, // Default timeout }) expect((vectorStore as any).qdrantUrl).toBe("http://localhost:6333") }) @@ -321,6 +359,7 @@ describe("QdrantVectorStore", () => { headers: { "User-Agent": "Roo-Code", }, + timeout: 30000, // Default timeout }) expect((vectorStore as any).qdrantUrl).toBe("http://localhost:6333") }) @@ -337,6 +376,7 @@ describe("QdrantVectorStore", () => { headers: { "User-Agent": "Roo-Code", }, + timeout: 30000, // Default timeout }) expect((vectorStore as any).qdrantUrl).toBe("http://invalid-url-format") }) @@ -359,6 +399,7 @@ describe("QdrantVectorStore", () => { headers: { "User-Agent": "Roo-Code", }, + timeout: 30000, // Default timeout }) expect((vectorStoreWithPrefix as any).qdrantUrl).toBe("http://localhost:6333/some/path") }) @@ -378,6 +419,7 @@ describe("QdrantVectorStore", () => { headers: { "User-Agent": "Roo-Code", }, + timeout: 30000, // Default timeout }) expect((vectorStoreWithoutPrefix as any).qdrantUrl).toBe("http://localhost:6333/") }) @@ -397,6 +439,7 @@ describe("QdrantVectorStore", () => { headers: { "User-Agent": "Roo-Code", }, + timeout: 30000, // Default timeout }) expect((vectorStoreWithHttpsPrefix as any).qdrantUrl).toBe("https://qdrant.ashbyfam.com/api") }) @@ -416,6 +459,7 @@ describe("QdrantVectorStore", () => { headers: { "User-Agent": "Roo-Code", }, + timeout: 30000, // Default timeout }) expect((vectorStoreWithTrailingSlash as any).qdrantUrl).toBe("http://localhost:6333/api/") }) @@ -435,6 +479,7 @@ describe("QdrantVectorStore", () => { headers: { "User-Agent": "Roo-Code", }, + timeout: 30000, // Default timeout }) expect((vectorStoreWithMultipleTrailingSlashes as any).qdrantUrl).toBe("http://localhost:6333/api///") }) @@ -454,6 +499,7 @@ describe("QdrantVectorStore", () => { headers: { "User-Agent": "Roo-Code", }, + timeout: 30000, // Default timeout }) expect((vectorStoreWithMultiSegment as any).qdrantUrl).toBe("http://localhost:6333/api/v1/qdrant") }) @@ -470,6 +516,7 @@ describe("QdrantVectorStore", () => { headers: { "User-Agent": "Roo-Code", }, + timeout: 30000, // Default timeout }) expect((vectorStoreComplex as any).qdrantUrl).toBe(complexUrl) }) @@ -489,6 +536,7 @@ describe("QdrantVectorStore", () => { headers: { "User-Agent": "Roo-Code", }, + timeout: 30000, // Default timeout }) expect((vectorStoreWithQueryParams as any).qdrantUrl).toBe( "http://localhost:6333/api/path?key=value#fragment", @@ -1263,4 +1311,131 @@ describe("QdrantVectorStore", () => { expect(callArgs.score_threshold).toBe(SEARCH_MIN_SCORE) }) }) + + describe("Socket error retry logic", () => { + it("should retry on UND_ERR_SOCKET errors", async () => { + const socketError = new Error("Socket error") + ;(socketError as any).code = "UND_ERR_SOCKET" + + // First call fails with socket error, second succeeds + mockQdrantClientInstance.upsert.mockRejectedValueOnce(socketError).mockResolvedValueOnce({} as any) + + const mockPoints = [ + { + id: "test-id", + vector: [0.1, 0.2, 0.3], + payload: { filePath: "test.ts", content: "test", startLine: 1, endLine: 1 }, + }, + ] + + await vectorStore.upsertPoints(mockPoints) + + expect(mockQdrantClientInstance.upsert).toHaveBeenCalledTimes(2) + }) + + it("should fail after max retries", async () => { + const socketError = new Error("Socket error") + ;(socketError as any).code = "UND_ERR_SOCKET" + + // All calls fail + mockQdrantClientInstance.upsert.mockRejectedValue(socketError) + vitest.spyOn(console, "error").mockImplementation(() => {}) + + const mockPoints = [ + { + id: "test-id", + vector: [0.1, 0.2, 0.3], + payload: { filePath: "test.ts", content: "test", startLine: 1, endLine: 1 }, + }, + ] + + await expect(vectorStore.upsertPoints(mockPoints)).rejects.toThrow(socketError) + + // Should retry 3 times (initial + 2 retries) + expect(mockQdrantClientInstance.upsert).toHaveBeenCalledTimes(3) + ;(console.error as any).mockRestore() + }) + + it("should not retry on non-socket errors", async () => { + const otherError = new Error("Other error") + mockQdrantClientInstance.upsert.mockRejectedValue(otherError) + vitest.spyOn(console, "error").mockImplementation(() => {}) + + const mockPoints = [ + { + id: "test-id", + vector: [0.1, 0.2, 0.3], + payload: { filePath: "test.ts", content: "test", startLine: 1, endLine: 1 }, + }, + ] + + await expect(vectorStore.upsertPoints(mockPoints)).rejects.toThrow(otherError) + + // Should not retry + expect(mockQdrantClientInstance.upsert).toHaveBeenCalledTimes(1) + ;(console.error as any).mockRestore() + }) + }) + + describe("Payload chunking", () => { + it("should chunk large payloads", async () => { + // Create points that exceed MAX_PAYLOAD_SIZE_BYTES when combined + const largeContent = "x".repeat(5 * 1024 * 1024) // 5MB content + const mockPoints = [ + { + id: "test-id-1", + vector: [0.1, 0.2, 0.3], + payload: { filePath: "test1.ts", content: largeContent, startLine: 1, endLine: 1 }, + }, + { + id: "test-id-2", + vector: [0.4, 0.5, 0.6], + payload: { filePath: "test2.ts", content: largeContent, startLine: 1, endLine: 1 }, + }, + { + id: "test-id-3", + vector: [0.7, 0.8, 0.9], + payload: { filePath: "test3.ts", content: largeContent, startLine: 1, endLine: 1 }, + }, + ] + + mockQdrantClientInstance.upsert.mockResolvedValue({} as any) + + await vectorStore.upsertPoints(mockPoints) + + // Should be called multiple times due to chunking + expect(mockQdrantClientInstance.upsert.mock.calls.length).toBeGreaterThan(1) + + // Each call should have points that fit within MAX_PAYLOAD_SIZE_BYTES + for (const call of mockQdrantClientInstance.upsert.mock.calls) { + const points = call[1].points + const payloadSize = JSON.stringify(points).length + expect(payloadSize).toBeLessThanOrEqual(10 * 1024 * 1024) // MAX_PAYLOAD_SIZE_BYTES + } + }) + + it("should handle single large point that exceeds max size", async () => { + // Create a single point that exceeds MAX_PAYLOAD_SIZE_BYTES + const veryLargeContent = "x".repeat(11 * 1024 * 1024) // 11MB content + const mockPoints = [ + { + id: "test-id-1", + vector: [0.1, 0.2, 0.3], + payload: { filePath: "test1.ts", content: veryLargeContent, startLine: 1, endLine: 1 }, + }, + ] + + mockQdrantClientInstance.upsert.mockResolvedValue({} as any) + vitest.spyOn(console, "warn").mockImplementation(() => {}) + + await vectorStore.upsertPoints(mockPoints) + + // Should still attempt to upsert the oversized point + expect(mockQdrantClientInstance.upsert).toHaveBeenCalledTimes(1) + expect(console.warn).toHaveBeenCalledWith( + expect.stringContaining("Single point exceeds maximum payload size"), + ) + ;(console.warn as any).mockRestore() + }) + }) }) diff --git a/src/services/code-index/vector-store/qdrant-client.ts b/src/services/code-index/vector-store/qdrant-client.ts index e975435b94..b46b67c192 100644 --- a/src/services/code-index/vector-store/qdrant-client.ts +++ b/src/services/code-index/vector-store/qdrant-client.ts @@ -3,7 +3,7 @@ import { createHash } from "crypto" import * as path from "path" import { IVectorStore } from "../interfaces/vector-store" import { Payload, VectorStoreSearchResult } from "../interfaces" -import { MAX_SEARCH_RESULTS, SEARCH_MIN_SCORE } from "../constants" +import { MAX_SEARCH_RESULTS, SEARCH_MIN_SCORE, MAX_PAYLOAD_SIZE_BYTES } from "../constants" /** * Qdrant implementation of the vector store interface @@ -21,8 +21,11 @@ export class QdrantVectorStore implements IVectorStore { * Creates a new Qdrant vector store * @param workspacePath Path to the workspace * @param url Optional URL to the Qdrant server + * @param vectorSize Size of the vectors + * @param apiKey Optional API key for authentication + * @param timeout Optional timeout in milliseconds */ - constructor(workspacePath: string, url: string, vectorSize: number, apiKey?: string) { + constructor(workspacePath: string, url: string, vectorSize: number, apiKey?: string, timeout?: number) { // Validate workspacePath is not empty if (!workspacePath || workspacePath.trim() === "") { throw new Error("Workspace path must not be empty") @@ -66,6 +69,8 @@ export class QdrantVectorStore implements IVectorStore { headers: { "User-Agent": "Roo-Code", }, + // Add timeout configuration + timeout: timeout || 30000, // Use provided timeout or default to 30 seconds }) } catch (urlError) { // If URL parsing fails, fall back to URL-based config @@ -76,6 +81,7 @@ export class QdrantVectorStore implements IVectorStore { headers: { "User-Agent": "Roo-Code", }, + timeout: timeout || 30000, // Use provided timeout or default to 30 seconds }) } @@ -216,6 +222,54 @@ export class QdrantVectorStore implements IVectorStore { } } + /** + * Estimates the payload size of points in bytes + * @param points Array of points to estimate + * @returns Estimated size in bytes + */ + private estimatePayloadSize(points: Array): number { + // Rough estimation of JSON payload size + return JSON.stringify(points).length + } + + /** + * Chunks points by payload size to avoid exceeding limits + * @param points Array of points to chunk + * @param maxSizeBytes Maximum size per chunk in bytes + * @returns Array of point chunks + */ + private chunkPointsBySize(points: Array, maxSizeBytes: number): Array> { + const chunks: Array> = [] + let currentChunk: Array = [] + let currentSize = 0 + + for (const point of points) { + const pointSize = this.estimatePayloadSize([point]) + + // Warn if a single point exceeds the max size + if (pointSize > maxSizeBytes) { + console.warn( + `[QdrantVectorStore] Single point exceeds maximum payload size (${pointSize} > ${maxSizeBytes}). It will be sent anyway but may fail.`, + ) + } + + if (currentSize + pointSize > maxSizeBytes && currentChunk.length > 0) { + chunks.push(currentChunk) + currentChunk = [] + currentSize = 0 + } + + currentChunk.push(point) + currentSize += pointSize + } + + if (currentChunk.length > 0) { + chunks.push(currentChunk) + } + + return chunks + } + /** * Upserts points into the vector store * @param points Array of points to upsert @@ -227,35 +281,57 @@ export class QdrantVectorStore implements IVectorStore { payload: Record }>, ): Promise { - try { - const processedPoints = points.map((point) => { - if (point.payload?.filePath) { - const segments = point.payload.filePath.split(path.sep).filter(Boolean) - const pathSegments = segments.reduce( - (acc: Record, segment: string, index: number) => { - acc[index.toString()] = segment - return acc - }, - {}, - ) - return { - ...point, - payload: { - ...point.payload, - pathSegments, - }, - } - } - return point - }) + const MAX_RETRIES = 3 + const INITIAL_DELAY = 1000 - await this.client.upsert(this.collectionName, { - points: processedPoints, - wait: true, - }) - } catch (error) { - console.error("Failed to upsert points:", error) - throw error + for (let attempt = 0; attempt < MAX_RETRIES; attempt++) { + try { + const processedPoints = points.map((point) => { + if (point.payload?.filePath) { + const segments = point.payload.filePath.split(path.sep).filter(Boolean) + const pathSegments = segments.reduce( + (acc: Record, segment: string, index: number) => { + acc[index.toString()] = segment + return acc + }, + {}, + ) + return { + ...point, + payload: { + ...point.payload, + pathSegments, + }, + } + } + return point + }) + + // Chunk by payload size + const chunks = this.chunkPointsBySize(processedPoints, MAX_PAYLOAD_SIZE_BYTES) + + for (const chunk of chunks) { + await this.client.upsert(this.collectionName, { + points: chunk, + wait: true, + }) + } + + return // Success + } catch (error: any) { + const isSocketError = error?.cause?.code === "UND_ERR_SOCKET" + const isLastAttempt = attempt === MAX_RETRIES - 1 + + if (!isSocketError || isLastAttempt) { + console.error("Failed to upsert points:", error) + throw error + } + + // Exponential backoff + const delay = INITIAL_DELAY * Math.pow(2, attempt) + console.warn(`Qdrant socket error, retrying in ${delay}ms (attempt ${attempt + 1}/${MAX_RETRIES})`) + await new Promise((resolve) => setTimeout(resolve, delay)) + } } }