diff --git a/src/i18n/locales/en/embeddings.json b/src/i18n/locales/en/embeddings.json index fc902cadc1..477330bfa0 100644 --- a/src/i18n/locales/en/embeddings.json +++ b/src/i18n/locales/en/embeddings.json @@ -61,6 +61,7 @@ "fileWatcherStopped": "File watcher stopped.", "failedDuringInitialScan": "Failed during initial scan: {{errorMessage}}", "unknownError": "Unknown error", - "indexingRequiresWorkspace": "Indexing requires an open workspace folder" + "indexingRequiresWorkspace": "Indexing requires an open workspace folder", + "qdrantNotAvailable": "{{errorMessage}}" } } diff --git a/src/services/code-index/__tests__/orchestrator.spec.ts b/src/services/code-index/__tests__/orchestrator.spec.ts new file mode 100644 index 0000000000..dc79582aba --- /dev/null +++ b/src/services/code-index/__tests__/orchestrator.spec.ts @@ -0,0 +1,333 @@ +import { describe, it, expect, beforeEach, afterEach, vi } from "vitest" +import * as vscode from "vscode" +import { CodeIndexOrchestrator } from "../orchestrator" +import { CodeIndexConfigManager } from "../config-manager" +import { CodeIndexStateManager } from "../state-manager" +import { IFileWatcher, IVectorStore } from "../interfaces" +import { DirectoryScanner } from "../processors" +import { CacheManager } from "../cache-manager" +import { t } from "../../../i18n" + +// Mock dependencies +vi.mock("vscode") +vi.mock("../config-manager") +vi.mock("../state-manager") +vi.mock("../processors") +vi.mock("../cache-manager") +vi.mock("@roo-code/telemetry", () => ({ + TelemetryService: { + instance: { + captureEvent: vi.fn(), + }, + }, + TelemetryEventName: { + CODE_INDEX_ERROR: "CODE_INDEX_ERROR", + }, +})) +vi.mock("../../../i18n", () => ({ + t: vi.fn((key: string, params?: any) => { + if (key === "embeddings:orchestrator.qdrantNotAvailable" && params?.errorMessage) { + return params.errorMessage + } + return key + }), +})) + +describe("CodeIndexOrchestrator", () => { + let orchestrator: CodeIndexOrchestrator + let mockConfigManager: any + let mockStateManager: any + let mockVectorStore: IVectorStore + let mockScanner: any + let mockFileWatcher: IFileWatcher + let mockCacheManager: any + + beforeEach(() => { + // Reset all mocks + vi.clearAllMocks() + vi.useFakeTimers() + + // Create mock instances + mockConfigManager = { + isFeatureConfigured: true, + } as any + + mockStateManager = { + setSystemState: vi.fn(), + state: "Standby", + reportBlockIndexingProgress: vi.fn(), + reportFileQueueProgress: vi.fn(), + } as any + + mockVectorStore = { + initialize: vi.fn().mockResolvedValue(false), + clearCollection: vi.fn().mockResolvedValue(undefined), + } as any + + mockScanner = { + scanDirectory: vi.fn().mockResolvedValue({ + stats: { + filesProcessed: 10, + blocksIndexed: 100, + }, + }), + } as any + + mockFileWatcher = { + initialize: vi.fn().mockResolvedValue(undefined), + dispose: vi.fn(), + onDidStartBatchProcessing: vi.fn().mockReturnValue({ dispose: vi.fn() }), + onBatchProgressUpdate: vi.fn().mockReturnValue({ dispose: vi.fn() }), + onDidFinishBatchProcessing: vi.fn().mockReturnValue({ dispose: vi.fn() }), + } as any + + mockCacheManager = { + clearCacheFile: vi.fn().mockResolvedValue(undefined), + } as any + + // Mock vscode workspace + ;(vscode.workspace as any).workspaceFolders = [ + { + uri: { fsPath: "/test/workspace" }, + }, + ] + + // Create orchestrator instance + orchestrator = new CodeIndexOrchestrator( + mockConfigManager as any, + mockStateManager as any, + "/test/workspace", + mockCacheManager as any, + mockVectorStore, + mockScanner as any, + mockFileWatcher, + ) + }) + + afterEach(() => { + vi.useRealTimers() + }) + + describe("Qdrant connection retry mechanism", () => { + it("should set up retry mechanism when Qdrant is not available", async () => { + // Mock Qdrant connection failure + const connectionError = new Error("Failed to connect to Qdrant vector database") + connectionError.message = "qdrantConnectionFailed" + mockVectorStore.initialize = vi.fn().mockRejectedValue(connectionError) + + // Start indexing + await orchestrator.startIndexing() + + // Verify error state is set + expect(mockStateManager.setSystemState).toHaveBeenCalledWith( + "Error", + expect.stringContaining("Qdrant service is not available"), + ) + + // Verify cache was NOT cleared + expect(mockCacheManager.clearCacheFile).not.toHaveBeenCalled() + }) + + it("should preserve cache when Qdrant connection fails", async () => { + // Mock Qdrant connection failure + const connectionError = new Error("connect ECONNREFUSED") + mockVectorStore.initialize = vi.fn().mockRejectedValue(connectionError) + + // Start indexing + await orchestrator.startIndexing() + + // Verify cache was NOT cleared + expect(mockCacheManager.clearCacheFile).not.toHaveBeenCalled() + + // Verify collection was NOT cleared + expect(mockVectorStore.clearCollection).not.toHaveBeenCalled() + }) + + it("should retry connection to Qdrant periodically", async () => { + // Mock initial connection failure + const connectionError = new Error("ECONNREFUSED") + mockVectorStore.initialize = vi + .fn() + .mockRejectedValueOnce(connectionError) + .mockRejectedValueOnce(connectionError) + .mockResolvedValueOnce(false) // Success on third attempt + + // Start indexing + await orchestrator.startIndexing() + + // Verify initial error state + expect(mockStateManager.setSystemState).toHaveBeenCalledWith( + "Error", + expect.stringContaining("Qdrant service is not available"), + ) + + // Fast-forward time to trigger first retry + await vi.advanceTimersByTimeAsync(30000) + + // Verify retry was attempted + expect(mockVectorStore.initialize).toHaveBeenCalledTimes(2) + + // Fast-forward time to trigger second retry (successful) + await vi.advanceTimersByTimeAsync(30000) + + // Verify successful reconnection + expect(mockVectorStore.initialize).toHaveBeenCalledTimes(3) + expect(mockStateManager.setSystemState).toHaveBeenCalledWith( + "Indexing", + "Qdrant connection restored. Resuming indexing...", + ) + }) + + it("should perform incremental indexing after Qdrant becomes available", async () => { + // Mock initial connection failure then success + const connectionError = new Error("ECONNREFUSED") + mockVectorStore.initialize = vi.fn().mockRejectedValueOnce(connectionError).mockResolvedValueOnce(false) // Success on retry, no new collection created + + // Start indexing + await orchestrator.startIndexing() + + // Fast-forward time to trigger retry + await vi.advanceTimersByTimeAsync(30000) + + // Verify incremental indexing was performed + expect(mockScanner.scanDirectory).toHaveBeenCalled() + expect(mockCacheManager.clearCacheFile).not.toHaveBeenCalled() + }) + + it("should clear cache only when new collection is created", async () => { + // Mock initial connection failure then success with new collection + const connectionError = new Error("ECONNREFUSED") + mockVectorStore.initialize = vi.fn().mockRejectedValueOnce(connectionError).mockResolvedValueOnce(true) // Success on retry, new collection created + + // Start indexing + await orchestrator.startIndexing() + + // Fast-forward time to trigger retry + await vi.advanceTimersByTimeAsync(30000) + + // Verify cache was cleared for new collection + expect(mockCacheManager.clearCacheFile).toHaveBeenCalledTimes(1) + }) + + it("should stop retrying after maximum attempts", async () => { + // Mock persistent connection failure + const connectionError = new Error("ECONNREFUSED") + mockVectorStore.initialize = vi.fn().mockRejectedValue(connectionError) + + // Start indexing + await orchestrator.startIndexing() + + // Fast-forward through all retry attempts + for (let i = 0; i < 10; i++) { + await vi.advanceTimersByTimeAsync(30000) + } + + // Verify maximum retry message + expect(mockStateManager.setSystemState).toHaveBeenLastCalledWith( + "Error", + "Maximum retry attempts reached. Please ensure Qdrant is running and restart indexing manually.", + ) + + // Verify no more retries after max + await vi.advanceTimersByTimeAsync(30000) + expect(mockVectorStore.initialize).toHaveBeenCalledTimes(11) // Initial + 10 retries + }) + + it("should clear retry timer when stopping watcher", async () => { + // Mock connection failure + const connectionError = new Error("ECONNREFUSED") + mockVectorStore.initialize = vi.fn().mockRejectedValue(connectionError) + + // Start indexing + await orchestrator.startIndexing() + + // Stop watcher + orchestrator.stopWatcher() + + // Fast-forward time + await vi.advanceTimersByTimeAsync(30000) + + // Verify no retry was attempted after stopping + expect(mockVectorStore.initialize).toHaveBeenCalledTimes(1) + }) + + it("should handle non-connection errors normally", async () => { + // Mock non-connection error + const otherError = new Error("Invalid configuration") + mockVectorStore.initialize = vi.fn().mockRejectedValue(otherError) + + // Start indexing + await orchestrator.startIndexing() + + // Verify normal error handling (cache cleared) + expect(mockCacheManager.clearCacheFile).toHaveBeenCalled() + expect(mockVectorStore.clearCollection).toHaveBeenCalled() + + // Verify no retry mechanism set up + await vi.advanceTimersByTimeAsync(30000) + expect(mockVectorStore.initialize).toHaveBeenCalledTimes(1) + }) + }) + + describe("startIndexing", () => { + it("should handle successful indexing flow", async () => { + // Mock successful initialization + mockVectorStore.initialize = vi.fn().mockResolvedValue(false) + + // Start indexing + await orchestrator.startIndexing() + + // Verify successful flow + expect(mockVectorStore.initialize).toHaveBeenCalled() + expect(mockScanner.scanDirectory).toHaveBeenCalled() + expect(mockFileWatcher.initialize).toHaveBeenCalled() + expect(mockStateManager.setSystemState).toHaveBeenCalledWith( + "Indexed", + "embeddings:orchestrator.fileWatcherStarted", + ) + }) + + it("should not proceed if no workspace folders", async () => { + // Mock no workspace + ;(vscode.workspace as any).workspaceFolders = [] + + // Start indexing + await orchestrator.startIndexing() + + // Verify early return + expect(mockVectorStore.initialize).not.toHaveBeenCalled() + expect(mockStateManager.setSystemState).toHaveBeenCalledWith( + "Error", + "embeddings:orchestrator.indexingRequiresWorkspace", + ) + }) + + it("should not proceed if feature not configured", async () => { + // Create a new mock with isFeatureConfigured set to false + const notConfiguredMockConfigManager = { + isFeatureConfigured: false, + } as any + + // Create a new orchestrator instance with the not configured mock + const notConfiguredOrchestrator = new CodeIndexOrchestrator( + notConfiguredMockConfigManager, + mockStateManager as any, + "/test/workspace", + mockCacheManager as any, + mockVectorStore, + mockScanner as any, + mockFileWatcher, + ) + + // Start indexing + await notConfiguredOrchestrator.startIndexing() + + // Verify early return + expect(mockVectorStore.initialize).not.toHaveBeenCalled() + expect(mockStateManager.setSystemState).toHaveBeenCalledWith( + "Standby", + "Missing configuration. Save your settings to start indexing.", + ) + }) + }) +}) diff --git a/src/services/code-index/orchestrator.ts b/src/services/code-index/orchestrator.ts index fbc4a24118..aa7d835701 100644 --- a/src/services/code-index/orchestrator.ts +++ b/src/services/code-index/orchestrator.ts @@ -15,6 +15,10 @@ import { t } from "../../i18n" export class CodeIndexOrchestrator { private _fileWatcherSubscriptions: vscode.Disposable[] = [] private _isProcessing: boolean = false + private _qdrantRetryTimer: NodeJS.Timeout | undefined + private _qdrantRetryCount: number = 0 + private readonly MAX_RETRY_COUNT = 10 + private readonly RETRY_INTERVAL_MS = 30000 // 30 seconds constructor( private readonly configManager: CodeIndexConfigManager, @@ -124,12 +128,54 @@ export class CodeIndexOrchestrator { this.stateManager.setSystemState("Indexing", "Initializing services...") try { - const collectionCreated = await this.vectorStore.initialize() + // Try to initialize the vector store with connection retry + let collectionCreated = false + let connectionError: Error | null = null + try { + collectionCreated = await this.vectorStore.initialize() + } catch (error: any) { + // Check if this is a connection error (Qdrant not available) + const errorMessage = error?.message || String(error) + if ( + errorMessage.includes("qdrantConnectionFailed") || + errorMessage.includes("ECONNREFUSED") || + errorMessage.includes("Failed to connect") || + errorMessage.includes("connect ECONNREFUSED") + ) { + connectionError = error as Error + console.warn( + "[CodeIndexOrchestrator] Qdrant connection failed, will attempt incremental indexing when available:", + errorMessage, + ) + // Don't throw here - continue with cache-based incremental indexing + } else { + // Other errors should still be thrown + throw error + } + } + + // Only clear cache if we successfully created a new collection + // This preserves the cache for incremental indexing when Qdrant comes back online if (collectionCreated) { await this.cacheManager.clearCacheFile() } + // If Qdrant is not available, we should not proceed with scanning + // Instead, set up monitoring for when it becomes available + if (connectionError) { + this.stateManager.setSystemState( + "Error", + t("embeddings:orchestrator.qdrantNotAvailable", { + errorMessage: + "Qdrant service is not available. Indexing will resume automatically when the service is restored.", + }), + ) + // Set up periodic retry for Qdrant connection + this._setupQdrantConnectionRetry() + return + } + this.stateManager.setSystemState("Indexing", "Services ready. Starting workspace scan...") let cumulativeBlocksIndexed = 0 @@ -210,25 +256,45 @@ export class CodeIndexOrchestrator { stack: error instanceof Error ? error.stack : undefined, location: "startIndexing", }) - try { - await this.vectorStore.clearCollection() - } catch (cleanupError) { - console.error("[CodeIndexOrchestrator] Failed to clean up after error:", cleanupError) - TelemetryService.instance.captureEvent(TelemetryEventName.CODE_INDEX_ERROR, { - error: cleanupError instanceof Error ? cleanupError.message : String(cleanupError), - stack: cleanupError instanceof Error ? cleanupError.stack : undefined, - location: "startIndexing.cleanup", - }) - } - await this.cacheManager.clearCacheFile() + // Check if this is a connection error - if so, don't clear the cache + const errorMessage = error?.message || String(error) + const isConnectionError = + errorMessage.includes("qdrantConnectionFailed") || + errorMessage.includes("ECONNREFUSED") || + errorMessage.includes("Failed to connect") || + errorMessage.includes("connect ECONNREFUSED") + + if (!isConnectionError) { + // Only clear collection and cache for non-connection errors + try { + await this.vectorStore.clearCollection() + } catch (cleanupError) { + console.error("[CodeIndexOrchestrator] Failed to clean up after error:", cleanupError) + TelemetryService.instance.captureEvent(TelemetryEventName.CODE_INDEX_ERROR, { + error: cleanupError instanceof Error ? cleanupError.message : String(cleanupError), + stack: cleanupError instanceof Error ? cleanupError.stack : undefined, + location: "startIndexing.cleanup", + }) + } + + await this.cacheManager.clearCacheFile() + } this.stateManager.setSystemState( "Error", - t("embeddings:orchestrator.failedDuringInitialScan", { - errorMessage: error.message || t("embeddings:orchestrator.unknownError"), - }), + isConnectionError + ? "Qdrant service is not available. Indexing will resume automatically when the service is restored." + : t("embeddings:orchestrator.failedDuringInitialScan", { + errorMessage: error.message || t("embeddings:orchestrator.unknownError"), + }), ) + + if (isConnectionError) { + // Set up periodic retry for Qdrant connection + this._setupQdrantConnectionRetry() + } + this.stopWatcher() } finally { this._isProcessing = false @@ -243,6 +309,12 @@ export class CodeIndexOrchestrator { this._fileWatcherSubscriptions.forEach((sub) => sub.dispose()) this._fileWatcherSubscriptions = [] + // Clear any pending retry timer + if (this._qdrantRetryTimer) { + clearTimeout(this._qdrantRetryTimer) + this._qdrantRetryTimer = undefined + } + if (this.stateManager.state !== "Error") { this.stateManager.setSystemState("Standby", t("embeddings:orchestrator.fileWatcherStopped")) } @@ -291,4 +363,138 @@ export class CodeIndexOrchestrator { public get state(): IndexingState { return this.stateManager.state } + + /** + * Sets up automatic retry mechanism for Qdrant connection + * Will periodically check if Qdrant becomes available and resume indexing + */ + private _setupQdrantConnectionRetry(): void { + // Clear any existing timer + if (this._qdrantRetryTimer) { + clearTimeout(this._qdrantRetryTimer) + } + + // Don't retry forever - have a reasonable limit + if (this._qdrantRetryCount >= this.MAX_RETRY_COUNT) { + console.warn( + `[CodeIndexOrchestrator] Max retry count (${this.MAX_RETRY_COUNT}) reached for Qdrant connection`, + ) + this.stateManager.setSystemState( + "Error", + "Maximum retry attempts reached. Please ensure Qdrant is running and restart indexing manually.", + ) + return + } + + this._qdrantRetryTimer = setTimeout(async () => { + console.log( + `[CodeIndexOrchestrator] Attempting to reconnect to Qdrant (attempt ${this._qdrantRetryCount + 1}/${this.MAX_RETRY_COUNT})...`, + ) + + try { + // Try to initialize the vector store + const collectionCreated = await this.vectorStore.initialize() + + // Success! Reset retry count and start indexing + console.log("[CodeIndexOrchestrator] Successfully reconnected to Qdrant!") + this._qdrantRetryCount = 0 + this._qdrantRetryTimer = undefined + + // Only clear cache if a new collection was created + if (collectionCreated) { + await this.cacheManager.clearCacheFile() + } + + // Resume indexing with incremental approach + this.stateManager.setSystemState("Indexing", "Qdrant connection restored. Resuming indexing...") + + // Start the indexing process + await this._performIncrementalIndexing() + } catch (error: any) { + // Still not available, schedule another retry + this._qdrantRetryCount++ + console.warn( + `[CodeIndexOrchestrator] Qdrant still not available, will retry in ${this.RETRY_INTERVAL_MS / 1000} seconds...`, + ) + + // Update status to show we're still retrying + this.stateManager.setSystemState( + "Error", + `Qdrant service not available. Retry attempt ${this._qdrantRetryCount}/${this.MAX_RETRY_COUNT}. Next retry in ${this.RETRY_INTERVAL_MS / 1000} seconds...`, + ) + + // Schedule next retry + this._setupQdrantConnectionRetry() + } + }, this.RETRY_INTERVAL_MS) + } + + /** + * Performs incremental indexing based on cached file hashes + * This is used when recovering from Qdrant connection failures + */ + private async _performIncrementalIndexing(): Promise { + if (this._isProcessing) { + console.warn("[CodeIndexOrchestrator] Already processing, skipping incremental indexing") + return + } + + this._isProcessing = true + + try { + this.stateManager.setSystemState("Indexing", "Performing incremental indexing...") + + let cumulativeBlocksIndexed = 0 + let cumulativeBlocksFoundSoFar = 0 + let batchErrors: Error[] = [] + + const handleFileParsed = (fileBlockCount: number) => { + cumulativeBlocksFoundSoFar += fileBlockCount + this.stateManager.reportBlockIndexingProgress(cumulativeBlocksIndexed, cumulativeBlocksFoundSoFar) + } + + const handleBlocksIndexed = (indexedCount: number) => { + cumulativeBlocksIndexed += indexedCount + this.stateManager.reportBlockIndexingProgress(cumulativeBlocksIndexed, cumulativeBlocksFoundSoFar) + } + + // Perform incremental scan using existing cache + const result = await this.scanner.scanDirectory( + this.workspacePath, + (batchError: Error) => { + console.error( + `[CodeIndexOrchestrator] Error during incremental scan batch: ${batchError.message}`, + batchError, + ) + batchErrors.push(batchError) + }, + handleBlocksIndexed, + handleFileParsed, + ) + + if (!result) { + throw new Error("Incremental scan failed") + } + + // Start the file watcher + await this._startWatcher() + + this.stateManager.setSystemState("Indexed", "Incremental indexing completed. Index up-to-date.") + } catch (error: any) { + console.error("[CodeIndexOrchestrator] Error during incremental indexing:", error) + TelemetryService.instance.captureEvent(TelemetryEventName.CODE_INDEX_ERROR, { + error: error instanceof Error ? error.message : String(error), + stack: error instanceof Error ? error.stack : undefined, + location: "_performIncrementalIndexing", + }) + + this.stateManager.setSystemState( + "Error", + `Failed during incremental indexing: ${error.message || "Unknown error"}`, + ) + this.stopWatcher() + } finally { + this._isProcessing = false + } + } }