Skip to content

Commit 16e5bcd

Browse files
WebMadWebMad
committed
fix(code-index): restore watcher events after restart
1 parent d351a15 commit 16e5bcd

3 files changed

Lines changed: 171 additions & 21 deletions

File tree

‎src/eslint-suppressions.json‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1396,7 +1396,7 @@
13961396
},
13971397
"services/code-index/processors/__tests__/file-watcher.spec.ts": {
13981398
"@typescript-eslint/no-explicit-any": {
1399-
"count": 25
1399+
"count": 22
14001400
}
14011401
},
14021402
"services/code-index/processors/__tests__/parser.spec.ts": {

‎src/services/code-index/processors/__tests__/file-watcher.spec.ts‎

Lines changed: 120 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -38,22 +38,25 @@ vi.mock("../parser", () => ({
3838
},
3939
}))
4040

41-
const createMockEventEmitter = () => {
42-
const listeners = new Set<(event: any) => void>()
41+
const createMockEventEmitter = <T>() => {
42+
let disposed = false
43+
const listeners = new Set<(event: T) => void>()
4344

4445
return {
45-
event: vi.fn((listener: (event: any) => void) => {
46-
listeners.add(listener)
46+
event: vi.fn((listener: (event: T) => void) => {
47+
if (!disposed) listeners.add(listener)
4748
return {
4849
dispose: () => listeners.delete(listener),
4950
}
5051
}),
51-
fire: vi.fn((event: any) => {
52+
fire: vi.fn((event: T) => {
53+
if (disposed) return
5254
for (const listener of listeners) {
5355
listener(event)
5456
}
5557
}),
5658
dispose: vi.fn(() => {
59+
disposed = true
5760
listeners.clear()
5861
}),
5962
}
@@ -88,6 +91,118 @@ vi.mock("vscode", () => ({
8891
}))
8992

9093
describe("FileWatcher", () => {
94+
it("does not deliver an old batch into subscriptions created after restart", async () => {
95+
await fileWatcher.initialize()
96+
let release!: () => void
97+
let notifyStarted!: () => void
98+
const started = new Promise<void>((resolve) => {
99+
notifyStarted = resolve
100+
})
101+
const blocked = new Promise<void>((resolve) => {
102+
release = resolve
103+
})
104+
vi.spyOn(fileWatcher, "processFile").mockImplementationOnce(async (path) => {
105+
notifyStarted()
106+
await blocked
107+
return { path, status: "skipped", reason: "test" }
108+
})
109+
await mockOnDidCreate(vscode.Uri.file("/mock/workspace/old.ts"))
110+
const processing = flushBatch()
111+
await started
112+
fileWatcher.dispose()
113+
await fileWatcher.initialize()
114+
const progress = vi.fn()
115+
const finished = vi.fn()
116+
fileWatcher.onBatchProgressUpdate(progress)
117+
fileWatcher.onDidFinishBatchProcessing(finished)
118+
release()
119+
await processing
120+
expect(progress).not.toHaveBeenCalled()
121+
expect(finished).not.toHaveBeenCalled()
122+
await mockOnDidCreate(vscode.Uri.file("/mock/workspace/new.ts"))
123+
await flushBatch()
124+
expect(progress).toHaveBeenCalled()
125+
expect(finished).toHaveBeenCalledOnce()
126+
})
127+
128+
it("restores all batch events after disposal and reinitialization", async () => {
129+
vi.mocked(vscode.workspace.createFileSystemWatcher).mockClear()
130+
await fileWatcher.initialize()
131+
const oldListener = vi.fn()
132+
fileWatcher.onDidFinishBatchProcessing(oldListener)
133+
fileWatcher.dispose()
134+
await fileWatcher.initialize()
135+
const started = vi.fn()
136+
const progress = vi.fn()
137+
const finished = vi.fn()
138+
fileWatcher.onDidStartBatchProcessing(started)
139+
fileWatcher.onBatchProgressUpdate(progress)
140+
fileWatcher.onDidFinishBatchProcessing(finished)
141+
await mockOnDidCreate(vscode.Uri.file("/mock/workspace/restarted.ts"))
142+
await flushBatch()
143+
expect(started).toHaveBeenCalledExactlyOnceWith(["/mock/workspace/restarted.ts"])
144+
expect(progress).toHaveBeenCalled()
145+
expect(finished).toHaveBeenCalledOnce()
146+
expect(oldListener).not.toHaveBeenCalled()
147+
expect(vscode.workspace.createFileSystemWatcher).toHaveBeenCalledTimes(2)
148+
})
149+
150+
it("does not create duplicate native watchers on repeated initialization", async () => {
151+
vi.mocked(vscode.workspace.createFileSystemWatcher).mockClear()
152+
await fileWatcher.initialize()
153+
await fileWatcher.initialize()
154+
expect(vscode.workspace.createFileSystemWatcher).toHaveBeenCalledOnce()
155+
})
156+
157+
it("discards queued events and the pending timer when disposed before a batch starts", async () => {
158+
await fileWatcher.initialize()
159+
await mockOnDidCreate(vscode.Uri.file("/mock/workspace/old.ts"))
160+
fileWatcher.dispose()
161+
fileWatcher.dispose()
162+
expect(mockWatcher.dispose).toHaveBeenCalledOnce()
163+
expect(vi.getTimerCount()).toBe(0)
164+
await fileWatcher.initialize()
165+
const started = vi.fn()
166+
fileWatcher.onDidStartBatchProcessing(started)
167+
await flushBatch()
168+
expect(started).not.toHaveBeenCalled()
169+
await mockOnDidCreate(vscode.Uri.file("/mock/workspace/new.ts"))
170+
await flushBatch()
171+
expect(started).toHaveBeenCalledExactlyOnceWith(["/mock/workspace/new.ts"])
172+
})
173+
174+
it("reports progress and preserves cached hashes when batch deletion fails", async () => {
175+
await fileWatcher.initialize()
176+
const error = new Error("deletion failed")
177+
mockVectorStore.deletePointsByMultipleFilePaths.mockRejectedValueOnce(error)
178+
const progress = vi.fn()
179+
const finished = vi.fn()
180+
fileWatcher.onBatchProgressUpdate(progress)
181+
fileWatcher.onDidFinishBatchProcessing(finished)
182+
const paths = ["/mock/workspace/first.ts", "/mock/workspace/second.ts"]
183+
184+
for (const path of paths) {
185+
await mockOnDidDelete(vscode.Uri.file(path))
186+
}
187+
await flushBatch()
188+
189+
expect(mockVectorStore.deletePointsByMultipleFilePaths).toHaveBeenCalledExactlyOnceWith(paths)
190+
expect(progress.mock.calls).toEqual([
191+
[{ processedInBatch: 0, totalInBatch: 2, currentFile: undefined }],
192+
[{ processedInBatch: 1, totalInBatch: 2, currentFile: paths[0] }],
193+
[{ processedInBatch: 2, totalInBatch: 2, currentFile: paths[1] }],
194+
[{ processedInBatch: 2, totalInBatch: 2 }],
195+
[{ processedInBatch: 0, totalInBatch: 0, currentFile: undefined }],
196+
])
197+
expect(finished).toHaveBeenCalledExactlyOnceWith({
198+
processedFiles: paths.map((path) => ({ path, status: "error", error })),
199+
batchError: error,
200+
})
201+
expect(mockCacheManager.deleteHash).not.toHaveBeenCalled()
202+
expect(mockCacheManager.updateHash).not.toHaveBeenCalled()
203+
expect(mockVectorStore.upsertPoints).not.toHaveBeenCalled()
204+
})
205+
91206
let fileWatcher: FileWatcher
92207
let mockWatcher: any
93208
let mockOnDidCreate: any

‎src/services/code-index/processors/file-watcher.ts‎

Lines changed: 50 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -41,28 +41,35 @@ export class FileWatcher implements IFileWatcher {
4141
private readonly FILE_PROCESSING_CONCURRENCY_LIMIT = 10
4242
private readonly batchSegmentThreshold: number
4343

44-
private readonly _onDidStartBatchProcessing = new vscode.EventEmitter<string[]>()
45-
private readonly _onBatchProgressUpdate = new vscode.EventEmitter<{
44+
private eventsDisposed = false
45+
private _onDidStartBatchProcessing = new vscode.EventEmitter<string[]>()
46+
private _onBatchProgressUpdate = new vscode.EventEmitter<{
4647
processedInBatch: number
4748
totalInBatch: number
4849
currentFile?: string
4950
}>()
50-
private readonly _onDidFinishBatchProcessing = new vscode.EventEmitter<BatchProcessingSummary>()
51+
private _onDidFinishBatchProcessing = new vscode.EventEmitter<BatchProcessingSummary>()
5152

5253
/**
5354
* Event emitted when a batch of files begins processing
5455
*/
55-
public readonly onDidStartBatchProcessing = this._onDidStartBatchProcessing.event
56+
public get onDidStartBatchProcessing() {
57+
return this._onDidStartBatchProcessing.event
58+
}
5659

5760
/**
5861
* Event emitted to report progress during batch processing
5962
*/
60-
public readonly onBatchProgressUpdate = this._onBatchProgressUpdate.event
63+
public get onBatchProgressUpdate() {
64+
return this._onBatchProgressUpdate.event
65+
}
6166

6267
/**
6368
* Event emitted when a batch of files has finished processing
6469
*/
65-
public readonly onDidFinishBatchProcessing = this._onDidFinishBatchProcessing.event
70+
public get onDidFinishBatchProcessing() {
71+
return this._onDidFinishBatchProcessing.event
72+
}
6673

6774
/**
6875
* Creates a new file watcher
@@ -106,6 +113,17 @@ export class FileWatcher implements IFileWatcher {
106113
* Initializes the file watcher
107114
*/
108115
async initialize(): Promise<void> {
116+
if (this.fileWatcher) return
117+
if (this.eventsDisposed) {
118+
this._onDidStartBatchProcessing = new vscode.EventEmitter<string[]>()
119+
this._onBatchProgressUpdate = new vscode.EventEmitter<{
120+
processedInBatch: number
121+
totalInBatch: number
122+
currentFile?: string
123+
}>()
124+
this._onDidFinishBatchProcessing = new vscode.EventEmitter<BatchProcessingSummary>()
125+
this.eventsDisposed = false
126+
}
109127
// Create file watcher
110128
const filePattern = new vscode.RelativePattern(
111129
this.workspacePath,
@@ -124,12 +142,15 @@ export class FileWatcher implements IFileWatcher {
124142
*/
125143
dispose(): void {
126144
this.fileWatcher?.dispose()
145+
this.fileWatcher = undefined
127146
if (this.batchProcessDebounceTimer) {
128147
clearTimeout(this.batchProcessDebounceTimer)
148+
this.batchProcessDebounceTimer = undefined
129149
}
130150
this._onDidStartBatchProcessing.dispose()
131151
this._onBatchProgressUpdate.dispose()
132152
this._onDidFinishBatchProcessing.dispose()
153+
this.eventsDisposed = true
133154
this.accumulatedEvents.clear()
134155
}
135156

@@ -181,10 +202,16 @@ export class FileWatcher implements IFileWatcher {
181202
const eventsToProcess = new Map(this.accumulatedEvents)
182203
this.accumulatedEvents.clear()
183204

205+
// Capture this session's emitters before notifying listeners or awaiting work.
206+
// Disposed emitters drop late events instead of forwarding them to a restarted session.
207+
const batchEvents = {
208+
progress: this._onBatchProgressUpdate,
209+
finished: this._onDidFinishBatchProcessing,
210+
}
184211
const filePathsInBatch = Array.from(eventsToProcess.keys())
185212
this._onDidStartBatchProcessing.fire(filePathsInBatch)
186213

187-
await this.processBatch(eventsToProcess)
214+
await this.processBatch(eventsToProcess, batchEvents)
188215
}
189216

190217
/**
@@ -197,6 +224,7 @@ export class FileWatcher implements IFileWatcher {
197224
totalFilesInBatch: number,
198225
pathsToExplicitlyDelete: string[],
199226
filesToUpsertDetails: Array<{ path: string; uri: vscode.Uri; originalType: "create" | "change" }>,
227+
progress: FileWatcher["_onBatchProgressUpdate"],
200228
): Promise<{ overallBatchError?: Error; clearedPaths: Set<string>; processedCount: number }> {
201229
let overallBatchError: Error | undefined
202230
const allPathsToClearFromDB = new Set<string>(pathsToExplicitlyDelete)
@@ -215,7 +243,7 @@ export class FileWatcher implements IFileWatcher {
215243
this.cacheManager.deleteHash(path)
216244
batchResults.push({ path, status: "success" })
217245
processedCountInBatch++
218-
this._onBatchProgressUpdate.fire({
246+
progress.fire({
219247
processedInBatch: processedCountInBatch,
220248
totalInBatch: totalFilesInBatch,
221249
currentFile: path,
@@ -238,7 +266,7 @@ export class FileWatcher implements IFileWatcher {
238266
for (const path of pathsToExplicitlyDelete) {
239267
batchResults.push({ path, status: "error", error: error as Error })
240268
processedCountInBatch++
241-
this._onBatchProgressUpdate.fire({
269+
progress.fire({
242270
processedInBatch: processedCountInBatch,
243271
totalInBatch: totalFilesInBatch,
244272
currentFile: path,
@@ -256,6 +284,7 @@ export class FileWatcher implements IFileWatcher {
256284
processedCountInBatch: number,
257285
totalFilesInBatch: number,
258286
pathsToExplicitlyDelete: string[],
287+
progress: FileWatcher["_onBatchProgressUpdate"],
259288
): Promise<{
260289
pointsForBatchUpsert: PointStruct[]
261290
successfullyProcessedForUpsert: Array<{ path: string; newHash?: string }>
@@ -269,7 +298,7 @@ export class FileWatcher implements IFileWatcher {
269298
const chunkToProcess = filesToProcessConcurrently.slice(i, i + this.FILE_PROCESSING_CONCURRENCY_LIMIT)
270299

271300
const chunkProcessingPromises = chunkToProcess.map(async (fileDetail) => {
272-
this._onBatchProgressUpdate.fire({
301+
progress.fire({
273302
processedInBatch: processedCountInBatch,
274303
totalInBatch: totalFilesInBatch,
275304
currentFile: fileDetail.path,
@@ -335,7 +364,7 @@ export class FileWatcher implements IFileWatcher {
335364
if (!pathsToExplicitlyDelete.includes(resultPath || "")) {
336365
processedCountInBatch++
337366
}
338-
this._onBatchProgressUpdate.fire({
367+
progress.fire({
339368
processedInBatch: processedCountInBatch,
340369
totalInBatch: totalFilesInBatch,
341370
currentFile: resultPath,
@@ -420,14 +449,18 @@ export class FileWatcher implements IFileWatcher {
420449

421450
private async processBatch(
422451
eventsToProcess: Map<string, { uri: vscode.Uri; type: "create" | "change" | "delete" }>,
452+
batchEvents: {
453+
progress: FileWatcher["_onBatchProgressUpdate"]
454+
finished: FileWatcher["_onDidFinishBatchProcessing"]
455+
},
423456
): Promise<void> {
424457
const batchResults: FileProcessingResult[] = []
425458
let processedCountInBatch = 0
426459
const totalFilesInBatch = eventsToProcess.size
427460
let overallBatchError: Error | undefined
428461

429462
// Initial progress update
430-
this._onBatchProgressUpdate.fire({
463+
batchEvents.progress.fire({
431464
processedInBatch: 0,
432465
totalInBatch: totalFilesInBatch,
433466
currentFile: undefined,
@@ -456,6 +489,7 @@ export class FileWatcher implements IFileWatcher {
456489
totalFilesInBatch,
457490
pathsToExplicitlyDelete,
458491
filesToUpsertDetails,
492+
batchEvents.progress,
459493
)
460494
overallBatchError = deletionError
461495
processedCountInBatch = deletionCount
@@ -471,6 +505,7 @@ export class FileWatcher implements IFileWatcher {
471505
processedCountInBatch,
472506
totalFilesInBatch,
473507
pathsToExplicitlyDelete,
508+
batchEvents.progress,
474509
)
475510
processedCountInBatch = upsertCount
476511

@@ -483,17 +518,17 @@ export class FileWatcher implements IFileWatcher {
483518
)
484519

485520
// Finalize
486-
this._onDidFinishBatchProcessing.fire({
521+
batchEvents.finished.fire({
487522
processedFiles: batchResults,
488523
batchError: overallBatchError,
489524
})
490-
this._onBatchProgressUpdate.fire({
525+
batchEvents.progress.fire({
491526
processedInBatch: totalFilesInBatch,
492527
totalInBatch: totalFilesInBatch,
493528
})
494529

495530
if (this.accumulatedEvents.size === 0) {
496-
this._onBatchProgressUpdate.fire({
531+
batchEvents.progress.fire({
497532
processedInBatch: 0,
498533
totalInBatch: 0,
499534
currentFile: undefined,

0 commit comments

Comments
 (0)