From 6fa7d048f963289bd1456631b688c6746c14675d Mon Sep 17 00:00:00 2001 From: matt wilkie Date: Mon, 5 Oct 2026 10:13:17 -0700 Subject: [PATCH 01/14] fix(indexing): serialize workspace index writes and preserve partial indexes Multiple sessions on the same workspace each ran their own scan: each treated the other's in-progress index as unusable, cleared it, and restarted from zero, leaving both sessions committing to the same LanceDB table (Kilo-Org/kilocode#14544). An interrupted index was likewise discarded instead of resumed (Kilo-Org/kilocode#14083). - writer-lock.ts: per-workspace lease lock under the indexing cache directory, heartbeat every 20s, reclaimed after 60s without a heartbeat or when the recorded local pid is gone - IVectorStore.state(): replaces hasIndexedData() with empty/partial/complete so an in-progress index is no longer indistinguishable from an absent one - orchestrator: take the writer lock before touching the store and clear the collection only when the store is empty; clearIndexData() takes the same lock - lancedb and qdrant stores implement state() Signed: _auto-free on behalf of @matt --- .changeset/single-writer-codebase-index.md | 5 + .../src/indexing/cache-manager.ts | 4 +- .../src/indexing/interfaces/vector-store.ts | 16 +- .../src/indexing/orchestrator.ts | 52 ++++++- .../vector-store/lancedb-vector-store.ts | 52 ++++--- .../indexing/vector-store/qdrant-client.ts | 59 +++---- .../kilo-indexing/src/indexing/writer-lock.ts | 145 ++++++++++++++++++ .../kilocode/indexing/orchestrator.test.ts | 99 ++++++++++-- .../indexing/processors/file-watcher.test.ts | 5 +- .../indexing/processors/scanner.test.ts | 5 +- .../kilocode/indexing/writer-lock.test.ts | 96 ++++++++++++ 11 files changed, 454 insertions(+), 84 deletions(-) create mode 100644 .changeset/single-writer-codebase-index.md create mode 100644 packages/kilo-indexing/src/indexing/writer-lock.ts create mode 100644 packages/kilo-indexing/test/kilocode/indexing/writer-lock.test.ts diff --git a/.changeset/single-writer-codebase-index.md b/.changeset/single-writer-codebase-index.md new file mode 100644 index 000000000000..d974ba3f53cb --- /dev/null +++ b/.changeset/single-writer-codebase-index.md @@ -0,0 +1,5 @@ +--- +"@kilocode/kilo-indexing": patch +--- + +Keep codebase indexing intact when several sessions share a workspace. Only one session now writes a workspace's index, and a session that finds a partially built index no longer discards it and starts over. \ No newline at end of file diff --git a/packages/kilo-indexing/src/indexing/cache-manager.ts b/packages/kilo-indexing/src/indexing/cache-manager.ts index ba5cd3735f17..6245264d3656 100644 --- a/packages/kilo-indexing/src/indexing/cache-manager.ts +++ b/packages/kilo-indexing/src/indexing/cache-manager.ts @@ -19,11 +19,11 @@ export class CacheManager implements ICacheManager { private saveTask = Promise.resolve() constructor( - private readonly cacheDirectory: string, + public readonly dir: string, private readonly workspacePath: string, ) { const hash = createHash("sha256").update(workspacePath).digest("hex") - this.cachePath = path.join(cacheDirectory, `roo-index-cache-${hash}.json`) + this.cachePath = path.join(dir, `roo-index-cache-${hash}.json`) } async initialize(): Promise { diff --git a/packages/kilo-indexing/src/indexing/interfaces/vector-store.ts b/packages/kilo-indexing/src/indexing/interfaces/vector-store.ts index 7712502f14a3..16df83caba35 100644 --- a/packages/kilo-indexing/src/indexing/interfaces/vector-store.ts +++ b/packages/kilo-indexing/src/indexing/interfaces/vector-store.ts @@ -70,10 +70,13 @@ export interface IVectorStore { collectionExists(): Promise /** - * Checks if the collection exists and has indexed points - * @returns Promise resolving to boolean indicating if the collection exists and has points + * Reports how far along the store is. `partial` means points exist but no + * successful scan finished, which is what a store owned by another live + * session looks like. Callers must treat `partial` as usable-but-untrusted + * and must never clear it on behalf of a session that does not own the + * writer lock. */ - hasIndexedData(): Promise + state(): Promise /** * Marks the indexing process as complete by storing metadata @@ -94,6 +97,13 @@ export interface VectorStoreSearchResult { payload?: Payload | null } +export type StoreStatus = "empty" | "partial" | "complete" + +export type StoreState = { + status: StoreStatus + points?: number +} + export interface Payload { filePath: string fileHash?: string diff --git a/packages/kilo-indexing/src/indexing/orchestrator.ts b/packages/kilo-indexing/src/indexing/orchestrator.ts index 9f4d376ecad4..c758f886e6ba 100644 --- a/packages/kilo-indexing/src/indexing/orchestrator.ts +++ b/packages/kilo-indexing/src/indexing/orchestrator.ts @@ -11,6 +11,7 @@ import type { IndexingTelemetryTrigger, } from "./interfaces/telemetry" import type { IVectorStore } from "./interfaces/vector-store" +import type { StoreState } from "./interfaces" import { DirectoryScanner } from "./processors" import type { CacheManager } from "./cache-manager" import type { Disposable } from "./runtime" @@ -18,9 +19,12 @@ import { Log } from "../util/log" import { sanitizeErrorMessage } from "./shared/validation-helpers" import { DEFAULT_VECTOR_STORE } from "./constants" import type { WorktreeOverlay } from "./worktree-overlay" +import { acquire, type Writer } from "./writer-lock" const log = Log.create({ service: "indexing-orchestrator" }) +const EMPTY: StoreState = { status: "empty" } + export class CodeIndexOrchestrator { private _fileWatcherSubscriptions: Disposable[] = [] private _isProcessing = false @@ -145,6 +149,26 @@ export class CodeIndexOrchestrator { return task } + /** + * Only one session may write a workspace's index. Without this, a second + * session starts a scan against a store the first one is still filling, + * clears it as unusable, and leaves both sessions committing to the same + * table. + */ + private async _acquire(): Promise { + const dir = this.cacheManager.dir + if (!dir) { + log.warn("no cache directory for the writer lock, proceeding unlocked", { + workspacePath: this.workspacePath, + }) + return { release: async () => {} } + } + const writer = await acquire(dir, this.workspacePath) + if (writer) return writer + log.info("writer lock is held elsewhere", { workspacePath: this.workspacePath }) + return undefined + } + private async runIndexing(trigger: IndexingTelemetryTrigger): Promise { log.info("indexing start requested", { workspacePath: this.workspacePath, @@ -176,6 +200,13 @@ export class CodeIndexOrchestrator { } this._cancelRequested = false + + const writer = await this._acquire() + if (!writer) { + this.stateManager.setSystemState("Standby", "Another session is indexing this workspace.") + return + } + this._isProcessing = true this.stateManager.setSystemState("Indexing", "Initializing services...") @@ -214,8 +245,11 @@ export class CodeIndexOrchestrator { }) } - const hasExistingData = this.overlay ? false : await this.vectorStore.hasIndexedData() - if (!this.overlay && !hasExistingData) { + const index = this.overlay ? EMPTY : await this.vectorStore.state() + // Only an empty store is safe to clear. A partial one belongs to a run + // that died or is still going, and clearing it throws away every point + // it managed to embed. + if (!this.overlay && index.status === "empty") { if (!collectionCreated) await this.vectorStore.clearCollection() await this.cacheManager.clearCacheFile() log.info("cleared indexing cache before full scan", { @@ -225,7 +259,7 @@ export class CodeIndexOrchestrator { } log.info("checked vector store indexed data", { workspacePath: this.workspacePath, - hasExistingData, + status: index.status, collectionCreated, }) @@ -234,7 +268,7 @@ export class CodeIndexOrchestrator { return } - mode = hasExistingData && !collectionCreated ? "incremental" : "full" + mode = index.status === "complete" && !collectionCreated ? "incremental" : "full" if (mode === "incremental") { log.info("collection has existing data, running incremental scan") @@ -245,7 +279,7 @@ export class CodeIndexOrchestrator { } else { log.info("running full scan", { workspacePath: this.workspacePath, - hasExistingData, + status: index.status, collectionCreated, }) this.stateManager.setSystemState("Indexing", "Services ready. Starting workspace scan...") @@ -267,6 +301,7 @@ export class CodeIndexOrchestrator { this.stopWatcher() } finally { this._isProcessing = false + await writer.release() log.info("indexing start flow finished", { workspacePath: this.workspacePath, state: this.stateManager.state, @@ -420,6 +455,12 @@ export class CodeIndexOrchestrator { } public async clearIndexData(): Promise { + const writer = await this._acquire() + if (!writer) { + this.stateManager.setSystemState("Standby", "Another session is indexing this workspace.") + return + } + this._isProcessing = true log.info("clearing index data", { workspacePath: this.workspacePath }) @@ -444,6 +485,7 @@ export class CodeIndexOrchestrator { } } finally { this._isProcessing = false + await writer.release() log.info("finished clearing index data", { workspacePath: this.workspacePath, state: this.stateManager.state, diff --git a/packages/kilo-indexing/src/indexing/vector-store/lancedb-vector-store.ts b/packages/kilo-indexing/src/indexing/vector-store/lancedb-vector-store.ts index f97363af68cd..e21e33896e00 100644 --- a/packages/kilo-indexing/src/indexing/vector-store/lancedb-vector-store.ts +++ b/packages/kilo-indexing/src/indexing/vector-store/lancedb-vector-store.ts @@ -2,7 +2,7 @@ import { createHash } from "crypto" import * as path from "path" import type { Connection, Table, VectorQuery } from "@lancedb/lancedb" import type { IVectorStore } from "../interfaces/vector-store" -import type { Payload, VectorStoreSearchResult } from "../interfaces" +import type { Payload, StoreState, VectorStoreSearchResult } from "../interfaces" import { DEFAULT_MAX_SEARCH_RESULTS, DEFAULT_SEARCH_MIN_SCORE } from "../constants" import fs from "fs" import { Log } from "../../util/log" @@ -590,31 +590,37 @@ export class LanceDBVectorStore implements IVectorStore { * Checks if the collection exists and has indexed points * @returns Promise resolving to boolean indicating if the collection exists and has points */ - async hasIndexedData(): Promise { - try { - const db = await this.getDb() - const table = await this.getTable() - const pointCount = await table.countRows() - if (pointCount === 0) { - log.info("LanceDB has no indexed data", { - workspacePath: this.workspacePath, - reason: "points_zero", - }) - return false - } - const metadataTable = await native(() => db.openTable(this.metadataTableName)) - const metadataResults = await metadataTable.query().where(`key = '${KEY.complete}'`).toArray() - const indexed = metadataResults.length > 0 ? String(metadataResults[0].value) === "true" : false - log.info("LanceDB indexing metadata evaluated", { + async state(): Promise { + const db = await this.getDb() + const tables = await db.tableNames() + if (!tables.includes(this.vectorTableName)) return { status: "empty" } + + const table = await native(() => db.openTable(this.vectorTableName)) + const points = await table.countRows() + if (points === 0) { + log.info("LanceDB has no indexed data", { workspacePath: this.workspacePath, - pointCount, - indexed, + reason: "points_zero", }) - return indexed - } catch (error) { - log.error("Failed to check if collection has data", { error }) - throw error + return { status: "empty", points } } + + if (!tables.includes(this.metadataTableName)) { + log.info("LanceDB index has points but no metadata table", { + workspacePath: this.workspacePath, + points, + }) + return { status: "partial", points } + } + + const rows = (await this._getMetadataValue(db, KEY.complete)) ?? "false" + const status = String(rows) === "true" ? "complete" : "partial" + log.info("LanceDB indexing metadata evaluated", { + workspacePath: this.workspacePath, + points, + status, + }) + return { status, points } } private async _upsertMetadata(metadataTable: Table, key: string, value: unknown): Promise { diff --git a/packages/kilo-indexing/src/indexing/vector-store/qdrant-client.ts b/packages/kilo-indexing/src/indexing/vector-store/qdrant-client.ts index ca3ee84641ef..ec17bfcd6894 100644 --- a/packages/kilo-indexing/src/indexing/vector-store/qdrant-client.ts +++ b/packages/kilo-indexing/src/indexing/vector-store/qdrant-client.ts @@ -2,7 +2,7 @@ import { QdrantClient, type Schemas } from "@qdrant/js-client-rest" import { createHash } from "crypto" import * as path from "path" import type { IVectorStore } from "../interfaces/vector-store" -import type { Payload, VectorStoreSearchResult } from "../interfaces" +import type { Payload, StoreState, VectorStoreSearchResult } from "../interfaces" import { DEFAULT_MAX_SEARCH_RESULTS, DEFAULT_SEARCH_MIN_SCORE } from "../constants" import { Log } from "../../util/log" import type { EmbeddingProfile } from "../embedding-profile" @@ -677,41 +677,32 @@ export class QdrantVectorStore implements IVectorStore { * Checks if the collection exists and has indexed points * @returns Promise resolving to boolean indicating if the collection exists and has points */ - async hasIndexedData(): Promise { - try { - const collectionInfo = await this.client.getCollection(this.collectionName) - // Check if the collection has any points indexed - const pointsCount = collectionInfo.points_count ?? 0 - if (pointsCount === 0) { - log.info("Qdrant collection has no indexed data", { - collection: this.collectionName, - reason: "points_zero", - }) - return false - } - - // Check if the indexing completion marker exists - const payload = await this.getMetadataPayload() - - // If marker exists, use it to determine completion status - if (payload) { - const indexed = payload[KEY.complete] === true - log.info("Qdrant indexing metadata evaluated", { - collection: this.collectionName, - pointsCount, - indexed, - }) - return indexed - } + async state(): Promise { + const collectionInfo = await this.client.getCollection(this.collectionName) + const points = collectionInfo.points_count ?? 0 + if (points === 0) { + log.info("Qdrant collection has no indexed data", { + collection: this.collectionName, + reason: "points_zero", + }) + return { status: "empty", points } + } - // Backward compatibility: No marker exists (old index or pre-marker version) - // Fall back to old logic - assume complete if collection has points - log.info("No indexing metadata marker found. Using backward compatibility mode (checking points_count > 0).") - return pointsCount > 0 - } catch (error) { - log.error("Failed to check if collection has data", { error }) - throw error + const payload = await this.getMetadataPayload() + if (payload) { + const status = payload[KEY.complete] === true ? "complete" : "partial" + log.info("Qdrant indexing metadata evaluated", { + collection: this.collectionName, + points, + status, + }) + return { status, points } } + + // Backward compatibility: no marker means an index from before the marker + // existed, which can only ever have been written by a finished scan. + log.info("No indexing metadata marker found. Using backward compatibility mode (checking points_count > 0).") + return { status: "complete", points } } /** diff --git a/packages/kilo-indexing/src/indexing/writer-lock.ts b/packages/kilo-indexing/src/indexing/writer-lock.ts new file mode 100644 index 000000000000..b7ca75a140ca --- /dev/null +++ b/packages/kilo-indexing/src/indexing/writer-lock.ts @@ -0,0 +1,145 @@ +import { createHash, randomUUID } from "crypto" +import fs from "fs/promises" +import os from "os" +import path from "path" +import { Log } from "../util/log" + +const log = Log.create({ service: "indexing-writer-lock" }) + +const STALE_MS = 60_000 +const BEAT_MS = Math.floor(STALE_MS / 3) + +export type Writer = { + release: () => Promise +} + +type Meta = { + token: string + pid: number + host: string +} + +function key(workspacePath: string): string { + return createHash("sha256").update(workspacePath).digest("hex").substring(0, 16) +} + +function code(err: unknown): string | undefined { + if (typeof err !== "object" || err === null || !("code" in err)) return undefined + return typeof err.code === "string" ? err.code : undefined +} + +function alive(pid: number): boolean { + try { + process.kill(pid, 0) + return true + } catch (err) { + return code(err) === "EPERM" + } +} + +async function age(file: string): Promise { + try { + return Date.now() - (await fs.stat(file)).mtimeMs + } catch (err) { + if (code(err) === "ENOENT") return undefined + throw err + } +} + +async function read(file: string): Promise { + try { + const parsed: unknown = JSON.parse(await fs.readFile(file, "utf-8")) + if (typeof parsed !== "object" || parsed === null) return undefined + if (!("token" in parsed) || !("pid" in parsed) || !("host" in parsed)) return undefined + return { token: String(parsed.token), pid: Number(parsed.pid), host: String(parsed.host) } + } catch (err) { + log.error("writer lock metadata unreadable", { err, file }) + return undefined + } +} + +async function claim(lockdir: string): Promise { + try { + await fs.mkdir(lockdir) + return true + } catch (err) { + if (code(err) === "EEXIST") return false + throw err + } +} + +/** + * A lock whose owner died leaves its directory behind forever, so ownership is + * decided by liveness rather than presence: a heartbeat that stopped, or a + * local pid that no longer exists, means the directory is safe to take. + */ +async function reclaimable(lockdir: string): Promise { + const beat = path.join(lockdir, "heartbeat") + const meta = path.join(lockdir, "meta.json") + + const beatAge = await age(beat) + if (beatAge !== undefined) return beatAge > STALE_MS + + const owner = await read(meta) + if (owner && owner.host === os.hostname() && !alive(owner.pid)) { + log.info("reclaiming writer lock from a dead local owner", { pid: owner.pid }) + return true + } + + const metaAge = await age(meta) + if (metaAge !== undefined) return metaAge > STALE_MS + + const dirAge = await age(lockdir) + return dirAge !== undefined && dirAge > STALE_MS +} + +/** + * Takes the writer lock for a workspace, or resolves undefined when another + * live session already holds it. Callers must not write to the vector store + * without holding this. + */ +export async function acquire(dir: string, workspacePath: string): Promise { + const root = path.join(dir, "locks") + const lockdir = path.join(root, key(workspacePath)) + + await fs.mkdir(root, { recursive: true }) + const made = await claim(lockdir) + if (!made) { + if (!(await reclaimable(lockdir))) { + log.info("writer lock held by another session", { workspacePath }) + return undefined + } + log.info("reclaiming stale writer lock", { workspacePath }) + await fs.rm(lockdir, { recursive: true, force: true }) + if (!(await claim(lockdir))) { + log.info("lost the race for a stale writer lock", { workspacePath }) + return undefined + } + } + + const token = randomUUID() + const owner = { token, pid: process.pid, host: os.hostname(), at: new Date().toISOString() } + await fs.writeFile(path.join(lockdir, "meta.json"), JSON.stringify(owner)) + await fs.writeFile(path.join(lockdir, "heartbeat"), "") + + const timer = setInterval(() => { + const now = new Date() + void fs + .utimes(path.join(lockdir, "heartbeat"), now, now) + .catch((err) => log.error("writer lock heartbeat failed", { err })) + }, BEAT_MS) + timer.unref?.() + + return { + release: async () => { + clearInterval(timer) + const current = await read(path.join(lockdir, "meta.json")) + if (current && current.token !== token) { + log.error("refusing to release a writer lock owned by another session", { workspacePath }) + return + } + await fs.rm(lockdir, { recursive: true, force: true }) + log.info("released writer lock", { workspacePath }) + }, + } +} diff --git a/packages/kilo-indexing/test/kilocode/indexing/orchestrator.test.ts b/packages/kilo-indexing/test/kilocode/indexing/orchestrator.test.ts index f2fc3f046de0..59c6da385829 100644 --- a/packages/kilo-indexing/test/kilocode/indexing/orchestrator.test.ts +++ b/packages/kilo-indexing/test/kilocode/indexing/orchestrator.test.ts @@ -1,4 +1,7 @@ import { describe, expect, test } from "bun:test" +import { mkdtemp, rm } from "fs/promises" +import { tmpdir } from "os" +import path from "path" import { CodeIndexConfigManager } from "../../../src/indexing/config-manager" import { CodeIndexOrchestrator } from "../../../src/indexing/orchestrator" import { CodeIndexStateManager } from "../../../src/indexing/state-manager" @@ -11,9 +14,12 @@ import type { IndexingTelemetryEvent, IVectorStore, PointStruct, + StoreState, + StoreStatus, VectorStoreSearchResult, } from "../../../src/indexing/interfaces" import { Emitter } from "../../../src/indexing/runtime" +import { acquire } from "../../../src/indexing/writer-lock" class Store { public clearCount = 0 @@ -23,7 +29,7 @@ class Store { public incompleteCount = 0 constructor( - private readonly existing: boolean, + private readonly status: StoreStatus, private readonly created = false, ) {} @@ -56,8 +62,8 @@ class Store { async collectionExists(): Promise { return true } - async hasIndexedData(): Promise { - return this.existing + async state(): Promise { + return { status: this.status } } async markIndexingComplete(): Promise { this.completeCount += 1 @@ -179,7 +185,7 @@ describe("CodeIndexOrchestrator telemetry", () => { { async clearCacheFile() {}, } as unknown as CacheManager, - new Store(false) as unknown as IVectorStore, + new Store("empty") as unknown as IVectorStore, new Scanner(3, 3, 6) as unknown as DirectoryScanner, new Watcher() as unknown as IFileWatcher, (event) => events.push(event), @@ -207,7 +213,7 @@ describe("CodeIndexOrchestrator telemetry", () => { { async clearCacheFile() {}, } as unknown as CacheManager, - new Store(true) as unknown as IVectorStore, + new Store("complete") as unknown as IVectorStore, new Scanner(2, 1, 2) as unknown as DirectoryScanner, new Watcher() as unknown as IFileWatcher, (event) => events.push(event), @@ -240,7 +246,7 @@ describe("CodeIndexOrchestrator telemetry", () => { new CodeIndexStateManager(), "/tmp/ws", { async clearCacheFile() {} } as unknown as CacheManager, - new Store(false) as unknown as IVectorStore, + new Store("empty") as unknown as IVectorStore, scanner, new Watcher() as unknown as IFileWatcher, ) @@ -258,7 +264,7 @@ describe("CodeIndexOrchestrator telemetry", () => { test("shutdown waits for an active scan before closing the store", async () => { const scanner = new BlockingScanner() - const store = new Store(false) + const store = new Store("empty") const orchestrator = new CodeIndexOrchestrator( createConfig(), new CodeIndexStateManager(), @@ -282,7 +288,7 @@ describe("CodeIndexOrchestrator telemetry", () => { test("preserves an unchanged index when an incremental scan is interrupted", async () => { const scanner = new BlockingScanner() - const store = new Store(true) + const store = new Store("complete") const orchestrator = new CodeIndexOrchestrator( createConfig(), new CodeIndexStateManager(), @@ -303,7 +309,7 @@ describe("CodeIndexOrchestrator telemetry", () => { expect(store.clearCount).toBe(0) }) - test("clears stale vectors and hashes before rebuilding an incomplete store", async () => { + test("clears stale vectors and hashes before rebuilding an empty store", async () => { const cache = { clears: 0, async clearCacheFile() { @@ -311,7 +317,7 @@ describe("CodeIndexOrchestrator telemetry", () => { }, async flush() {}, } - const store = new Store(false, false) + const store = new Store("empty", false) const orchestrator = new CodeIndexOrchestrator( createConfig(), new CodeIndexStateManager(), @@ -329,6 +335,73 @@ describe("CodeIndexOrchestrator telemetry", () => { expect(orchestrator.state).toBe("Indexed") }) + test("does not clear a partially built index", async () => { + const cache = { + clears: 0, + async clearCacheFile() { + this.clears += 1 + }, + async flush() {}, + } + const store = new Store("partial", false) + const events: IndexingTelemetryEvent[] = [] + const orchestrator = new CodeIndexOrchestrator( + createConfig(), + new CodeIndexStateManager(), + "/tmp/ws", + cache as unknown as CacheManager, + store as unknown as IVectorStore, + new Scanner(2, 2, 4) as unknown as DirectoryScanner, + new Watcher() as unknown as IFileWatcher, + (event) => events.push(event), + ) + + await orchestrator.startIndexing("background") + + const completed = events.find( + (event): event is Extract => event.type === "completed", + ) + expect(store.clearCount).toBe(0) + expect(cache.clears).toBe(0) + expect(completed?.mode).toBe("full") + expect(orchestrator.state).toBe("Indexed") + }) + + test("stands by while another session holds the writer lock", async () => { + const dir = await mkdtemp(path.join(tmpdir(), "indexing-lock-")) + const held = await acquire(dir, "/tmp/ws") + expect(held).toBeDefined() + + const store = new Store("partial", false) + let scanned = false + const scanner = new Scanner(1, 1, 1) as unknown as DirectoryScanner + const scan = scanner.scanDirectory.bind(scanner) + scanner.scanDirectory = async (...args: Parameters) => { + scanned = true + return scan(...args) + } + const orchestrator = new CodeIndexOrchestrator( + createConfig(), + new CodeIndexStateManager(), + "/tmp/ws", + { dir, async clearCacheFile() {}, async flush() {} } as unknown as CacheManager, + store as unknown as IVectorStore, + scanner, + new Watcher() as unknown as IFileWatcher, + ) + + await orchestrator.startIndexing("background") + + expect(scanned).toBe(false) + expect(store.clearCount).toBe(0) + expect(store.deleteCount).toBe(0) + expect(store.incompleteCount).toBe(0) + expect(orchestrator.state).toBe("Standby") + + await held?.release() + await rm(dir, { recursive: true, force: true }) + }) + test("does not clear data when index completeness cannot be read", async () => { const cache = { clears: 0, @@ -336,8 +409,8 @@ describe("CodeIndexOrchestrator telemetry", () => { this.clears += 1 }, } - const store = new Store(true, false) - store.hasIndexedData = async () => { + const store = new Store("complete", false) + store.state = async () => { throw new Error("metadata unavailable") } const orchestrator = new CodeIndexOrchestrator( @@ -365,7 +438,7 @@ describe("CodeIndexOrchestrator telemetry", () => { this.clears += 1 }, } - const store = new Store(true) + const store = new Store("complete") const orchestrator = new CodeIndexOrchestrator( createConfig(), new CodeIndexStateManager(), diff --git a/packages/kilo-indexing/test/kilocode/indexing/processors/file-watcher.test.ts b/packages/kilo-indexing/test/kilocode/indexing/processors/file-watcher.test.ts index 2402a00d6855..8ab192fa47c1 100644 --- a/packages/kilo-indexing/test/kilocode/indexing/processors/file-watcher.test.ts +++ b/packages/kilo-indexing/test/kilocode/indexing/processors/file-watcher.test.ts @@ -11,6 +11,7 @@ import type { IndexingTelemetryEvent, IVectorStore, PointStruct, + StoreState, VectorStoreSearchResult, } from "../../../../src/indexing/interfaces" import { @@ -76,8 +77,8 @@ class RetryStore implements IVectorStore { async collectionExists(): Promise { return true } - async hasIndexedData(): Promise { - return false + async state(): Promise { + return { status: "empty" } } async markIndexingComplete(): Promise {} async markIndexingIncomplete(): Promise {} diff --git a/packages/kilo-indexing/test/kilocode/indexing/processors/scanner.test.ts b/packages/kilo-indexing/test/kilocode/indexing/processors/scanner.test.ts index ea2e75a2758c..66fad5cf9d72 100644 --- a/packages/kilo-indexing/test/kilocode/indexing/processors/scanner.test.ts +++ b/packages/kilo-indexing/test/kilocode/indexing/processors/scanner.test.ts @@ -12,6 +12,7 @@ import type { IndexingTelemetryEvent, IVectorStore, PointStruct, + StoreState, VectorStoreSearchResult, } from "../../../../src/indexing/interfaces" import { loadIgnore } from "../../../../src/indexing/shared/load-ignore" @@ -138,8 +139,8 @@ class Store implements IVectorStore { return true } - public async hasIndexedData(): Promise { - return false + public async state(): Promise { + return { status: "empty" } } public async markIndexingComplete(): Promise {} diff --git a/packages/kilo-indexing/test/kilocode/indexing/writer-lock.test.ts b/packages/kilo-indexing/test/kilocode/indexing/writer-lock.test.ts new file mode 100644 index 000000000000..7e27716595d9 --- /dev/null +++ b/packages/kilo-indexing/test/kilocode/indexing/writer-lock.test.ts @@ -0,0 +1,96 @@ +import { afterEach, beforeEach, describe, expect, test } from "bun:test" +import { mkdir, mkdtemp, readFile, rm, utimes, writeFile } from "fs/promises" +import { hostname, tmpdir } from "os" +import path from "path" +import { createHash } from "crypto" +import { acquire } from "../../../src/indexing/writer-lock" + +const WS = "/tmp/workspace-under-test" + +let dir: string + +function slot(): string { + return path.join(dir, "locks", createHash("sha256").update(WS).digest("hex").substring(0, 16)) +} + +async function meta(): Promise<{ token: string; pid: number; host: string; at: string }> { + return JSON.parse(await readFile(path.join(slot(), "meta.json"), "utf-8")) +} + +async function plant(owner: { token: string; pid: number; host: string }, beat = false): Promise { + await mkdir(slot(), { recursive: true }) + await writeFile(path.join(slot(), "meta.json"), JSON.stringify({ ...owner, at: new Date().toISOString() })) + if (!beat) return + await writeFile(path.join(slot(), "heartbeat"), "") + const when = new Date(Date.now() - 120_000) + await utimes(path.join(slot(), "heartbeat"), when, when) +} + +beforeEach(async () => { + dir = await mkdtemp(path.join(tmpdir(), "writer-lock-")) +}) + +afterEach(async () => { + await rm(dir, { recursive: true, force: true }) +}) + +describe("writer lock", () => { + test("grants the lock once and refuses a second holder", async () => { + const first = await acquire(dir, WS) + expect(first).toBeDefined() + expect(await acquire(dir, WS)).toBeUndefined() + await first?.release() + }) + + test("grants the lock again after release", async () => { + const first = await acquire(dir, WS) + await first?.release() + expect(await acquire(dir, WS)).toBeDefined() + }) + + test("scopes the lock to the workspace", async () => { + const first = await acquire(dir, WS) + expect(await acquire(dir, "/tmp/some-other-workspace")).toBeDefined() + await first?.release() + }) + + test("records the owning process", async () => { + const writer = await acquire(dir, WS) + expect((await meta()).pid).toBe(process.pid) + await writer?.release() + }) + + test("takes over a lock whose heartbeat stopped", async () => { + await plant({ token: "dead", pid: process.pid, host: "elsewhere" }, true) + + const writer = await acquire(dir, WS) + expect(writer).toBeDefined() + expect((await meta()).token).not.toBe("dead") + await writer?.release() + }) + + test("takes over a lock whose local owner is gone", async () => { + await plant({ token: "dead", pid: 2 ** 30, host: hostname() }) + + const writer = await acquire(dir, WS) + expect(writer).toBeDefined() + expect((await meta()).token).not.toBe("dead") + await writer?.release() + }) + + test("leaves a fresh lock held by a live owner alone", async () => { + await plant({ token: "live", pid: process.pid, host: hostname() }) + expect(await acquire(dir, WS)).toBeUndefined() + }) + + test("refuses to release a lock that was taken over", async () => { + const first = await acquire(dir, WS) + await rm(slot(), { recursive: true, force: true }) + const second = await acquire(dir, WS) + const token = (await meta()).token + + await first?.release() + expect((await meta()).token).toBe(token) + await second?.release() + }) +}) From bce5b0a7fd095d4cc22a76be2f8bb66b5d75ed31 Mon Sep 17 00:00:00 2001 From: matt wilkie Date: Tue, 6 Oct 2026 13:53:26 -0700 Subject: [PATCH 02/14] chore(changeset): mark single-writer change as minor for breaking interface change --- .changeset/single-writer-codebase-index.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.changeset/single-writer-codebase-index.md b/.changeset/single-writer-codebase-index.md index d974ba3f53cb..7cd811e940b4 100644 --- a/.changeset/single-writer-codebase-index.md +++ b/.changeset/single-writer-codebase-index.md @@ -1,5 +1,5 @@ --- -"@kilocode/kilo-indexing": patch +"@kilocode/kilo-indexing": minor --- Keep codebase indexing intact when several sessions share a workspace. Only one session now writes a workspace's index, and a session that finds a partially built index no longer discards it and starts over. \ No newline at end of file From 48f5dba6698c2de6dcdc777e40f5383ac3678ce3 Mon Sep 17 00:00:00 2001 From: matt wilkie Date: Sat, 10 Oct 2026 07:14:58 -0700 Subject: [PATCH 03/14] docs(indexing): correct state() doc comment on vector stores The state() method kept the hasIndexedData() doc comment claiming a boolean return; it returns StoreState. Signed: _kilo-auto/free on behalf of maphew --- .../src/indexing/vector-store/lancedb-vector-store.ts | 4 ++-- .../kilo-indexing/src/indexing/vector-store/qdrant-client.ts | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/packages/kilo-indexing/src/indexing/vector-store/lancedb-vector-store.ts b/packages/kilo-indexing/src/indexing/vector-store/lancedb-vector-store.ts index e21e33896e00..c3ad2ba62a85 100644 --- a/packages/kilo-indexing/src/indexing/vector-store/lancedb-vector-store.ts +++ b/packages/kilo-indexing/src/indexing/vector-store/lancedb-vector-store.ts @@ -587,8 +587,8 @@ export class LanceDBVectorStore implements IVectorStore { } /** - * Checks if the collection exists and has indexed points - * @returns Promise resolving to boolean indicating if the collection exists and has points + * Reports how far along the store is: empty, partially built, or + * complete. */ async state(): Promise { const db = await this.getDb() diff --git a/packages/kilo-indexing/src/indexing/vector-store/qdrant-client.ts b/packages/kilo-indexing/src/indexing/vector-store/qdrant-client.ts index ec17bfcd6894..196d0b983ce7 100644 --- a/packages/kilo-indexing/src/indexing/vector-store/qdrant-client.ts +++ b/packages/kilo-indexing/src/indexing/vector-store/qdrant-client.ts @@ -674,8 +674,8 @@ export class QdrantVectorStore implements IVectorStore { } /** - * Checks if the collection exists and has indexed points - * @returns Promise resolving to boolean indicating if the collection exists and has points + * Reports how far along the store is: empty, partially built, or + * complete. */ async state(): Promise { const collectionInfo = await this.client.getCollection(this.collectionName) From dc3a7b20d06488836dc1297d98f932e4f3123f20 Mon Sep 17 00:00:00 2001 From: maphew <486200+maphew@users.noreply.github.com> Date: Sun, 11 Oct 2026 05:48:02 +0000 Subject: [PATCH 04/14] fix(indexing): harden single-writer lock lifecycle and contention --- .../kilo-indexing/src/indexing/manager.ts | 7 ++ .../src/indexing/orchestrator.ts | 61 +++++++++++---- .../indexing/vector-store/qdrant-client.ts | 15 ++-- .../kilo-indexing/src/indexing/writer-lock.ts | 78 +++++++++++++++---- .../kilocode/indexing/orchestrator.test.ts | 40 ++++++++++ .../vector-store/qdrant-client.test.ts | 20 +++++ .../kilocode/indexing/writer-lock.test.ts | 28 ++++++- 7 files changed, 215 insertions(+), 34 deletions(-) diff --git a/packages/kilo-indexing/src/indexing/manager.ts b/packages/kilo-indexing/src/indexing/manager.ts index dcaed039b19c..05f8fdfc49ff 100644 --- a/packages/kilo-indexing/src/indexing/manager.ts +++ b/packages/kilo-indexing/src/indexing/manager.ts @@ -160,6 +160,13 @@ export class CodeIndexManager { if (!this.isFeatureEnabled || !this.isFeatureConfigured) return if (this._retryTask || this._isRecoveringFromError) return + // Lock contention is not failure: another live session holds the writer + // lock, so retrying with backoff only spins while the holder watches. + // The second session retries on its own triggers (settings, restart, + // watcher events); a retry here would also recreate services, tearing + // down the loser's search path for no gain. + if (event.error.includes("Another session is indexing this workspace")) return + if (this._retryAttempt >= this._retryMaxAttempts) { log.warn("indexing recovery retries exhausted", { workspacePath: this.workspacePath, diff --git a/packages/kilo-indexing/src/indexing/orchestrator.ts b/packages/kilo-indexing/src/indexing/orchestrator.ts index c758f886e6ba..d4572bb377f2 100644 --- a/packages/kilo-indexing/src/indexing/orchestrator.ts +++ b/packages/kilo-indexing/src/indexing/orchestrator.ts @@ -201,20 +201,34 @@ export class CodeIndexOrchestrator { this._cancelRequested = false - const writer = await this._acquire() - if (!writer) { - this.stateManager.setSystemState("Standby", "Another session is indexing this workspace.") - return - } - this._isProcessing = true this.stateManager.setSystemState("Indexing", "Initializing services...") let started = false let source: IndexingTelemetrySource = "watcher" let mode: IndexingTelemetryMode | undefined + let writer: Writer | undefined try { + // Acquired inside the try: a throw between acquire and the body (e.g. + // from state setup below) would otherwise skip the release and keep the + // lock fresh for STALE_MS, blocking every other session. + writer = await this._acquire() + if (!writer) { + // Contention is worth observing, not a silent park: the manager + // surfaces start events as activity, so emit an error event the manager + // can retry or display rather than leaving the session in Standby with + // no record of the missed run. + this.emitError( + "orchestrator:startIndexing", + new Error("Another session is indexing this workspace."), + "scan", + trigger, + ) + this.stateManager.setSystemState("Standby", "Another session is indexing this workspace.") + return + } + this.overlay?.prepare() await this._startWatcher() @@ -301,7 +315,7 @@ export class CodeIndexOrchestrator { this.stopWatcher() } finally { this._isProcessing = false - await writer.release() + await writer?.release() log.info("indexing start flow finished", { workspacePath: this.workspacePath, state: this.stateManager.state, @@ -394,7 +408,10 @@ export class CodeIndexOrchestrator { this.overlay?.reconcile(this.cacheManager.getAllHashes()) await this.cacheManager.flush?.() await this.vectorStore.markIndexingComplete() - await this.vectorStore.close?.() + // The scan finished, but this session's file watcher still writes the + // store for every file change. Releasing the writer lock here would let a + // second session acquire it and scan concurrently — the corruption the + // lock targets — so the lock stays held until shutdown or clearIndexData. this.fileWatcher.setCollecting(true) this.stateManager.setSystemState("Indexed", "File watcher started. Index up-to-date.") log.info("workspace scan finalized", { @@ -420,13 +437,20 @@ export class CodeIndexOrchestrator { public async shutdown(): Promise { this._cancelRequested = true this.scanner.cancel() + // Stop the session's own watcher writes before releasing the writer lock: + // _active's release happens in runIndexing's finally, so the store must be + // quiet by then rather than still receiving file-watcher batches. + if (this.fileWatcher.shutdown) await this.fileWatcher.shutdown() + else this.fileWatcher.dispose() this.fileWatcher.setCollecting(false) + // Close the store before the writer lock releases in _active's finally: + // _active runs runIndexing, whose finally releases the lock, so closing + // here hands the next session a closed connection rather than one whose + // close is still in flight. + await this.vectorStore.close?.() await this._active for (const sub of this._fileWatcherSubscriptions) sub.dispose() this._fileWatcherSubscriptions = [] - if (this.fileWatcher.shutdown) await this.fileWatcher.shutdown() - else this.fileWatcher.dispose() - await this.vectorStore.close?.() this._isProcessing = false } @@ -448,7 +472,10 @@ export class CodeIndexOrchestrator { log.info("cancelling indexing", { workspacePath: this.workspacePath }) this._cancelRequested = true this.scanner.cancel() - this.stopWatcher() + // Keep the file watcher alive: cancel only aborts the scan, and the + // session still holds the writer lock, so watcher batches continue to own + // the store. Disposing here (via stopWatcher) would strand _active's + // finally with a disposed watcher while file changes keep arriving. this.stateManager.setSystemState("Standby", "Indexing cancelled.") this._isProcessing = false log.info("indexing cancelled", { workspacePath: this.workspacePath }) @@ -457,7 +484,15 @@ export class CodeIndexOrchestrator { public async clearIndexData(): Promise { const writer = await this._acquire() if (!writer) { - this.stateManager.setSystemState("Standby", "Another session is indexing this workspace.") + // A user-requested clear must not vanish silently when another session + // holds the lock: record contention as an error event so the manager can + // retry or display it, instead of reporting Standby success. + this.emitError( + "orchestrator:startIndexing", + new Error("Another session is indexing this workspace; index data was not cleared."), + "scan", + ) + this.stateManager.setSystemState("Error", "Another session is indexing this workspace; index data was not cleared.") return } diff --git a/packages/kilo-indexing/src/indexing/vector-store/qdrant-client.ts b/packages/kilo-indexing/src/indexing/vector-store/qdrant-client.ts index 196d0b983ce7..581827ea2e51 100644 --- a/packages/kilo-indexing/src/indexing/vector-store/qdrant-client.ts +++ b/packages/kilo-indexing/src/indexing/vector-store/qdrant-client.ts @@ -680,12 +680,17 @@ export class QdrantVectorStore implements IVectorStore { async state(): Promise { const collectionInfo = await this.client.getCollection(this.collectionName) const points = collectionInfo.points_count ?? 0 - if (points === 0) { + // The completion marker is itself a point, so subtract it before judging + // emptiness: otherwise a scanned-then-cleared collection reports + // points_count 1 and `empty` is unreachable after any scan. + const marker = (await this.getMetadataPayload()) ? 1 : 0 + const indexed = Math.max(0, points - marker) + if (indexed === 0) { log.info("Qdrant collection has no indexed data", { collection: this.collectionName, reason: "points_zero", }) - return { status: "empty", points } + return { status: "empty", points: indexed } } const payload = await this.getMetadataPayload() @@ -693,16 +698,16 @@ export class QdrantVectorStore implements IVectorStore { const status = payload[KEY.complete] === true ? "complete" : "partial" log.info("Qdrant indexing metadata evaluated", { collection: this.collectionName, - points, + points: indexed, status, }) - return { status, points } + return { status, points: indexed } } // Backward compatibility: no marker means an index from before the marker // existed, which can only ever have been written by a finished scan. log.info("No indexing metadata marker found. Using backward compatibility mode (checking points_count > 0).") - return { status: "complete", points } + return { status: "complete", points: indexed } } /** diff --git a/packages/kilo-indexing/src/indexing/writer-lock.ts b/packages/kilo-indexing/src/indexing/writer-lock.ts index b7ca75a140ca..3889541958ed 100644 --- a/packages/kilo-indexing/src/indexing/writer-lock.ts +++ b/packages/kilo-indexing/src/indexing/writer-lock.ts @@ -72,25 +72,35 @@ async function claim(lockdir: string): Promise { * A lock whose owner died leaves its directory behind forever, so ownership is * decided by liveness rather than presence: a heartbeat that stopped, or a * local pid that no longer exists, means the directory is safe to take. + * + * A vanished directory reports as reclaimable: without this, a session that + * lost the race on `claim` would treat its own missing directory (deleted by + * the winner's reclaim just before its claim) as held-forever and wedge in + * Standby instead of retrying the claim. */ async function reclaimable(lockdir: string): Promise { const beat = path.join(lockdir, "heartbeat") const meta = path.join(lockdir, "meta.json") - const beatAge = await age(beat) - if (beatAge !== undefined) return beatAge > STALE_MS - + // Check the dead-local-owner fast path before heartbeat freshness: a dead + // pid cannot refresh its heartbeat, so waiting out STALE_MS on a fresh file + // only delays an inevitable reclaim. (Pid reuse is safe: a reused pid reads + // alive and falls through to the age checks.) const owner = await read(meta) if (owner && owner.host === os.hostname() && !alive(owner.pid)) { log.info("reclaiming writer lock from a dead local owner", { pid: owner.pid }) return true } + const beatAge = await age(beat) + if (beatAge !== undefined) return beatAge > STALE_MS + const metaAge = await age(meta) if (metaAge !== undefined) return metaAge > STALE_MS const dirAge = await age(lockdir) - return dirAge !== undefined && dirAge > STALE_MS + if (dirAge === undefined) return true + return dirAge > STALE_MS } /** @@ -103,15 +113,18 @@ export async function acquire(dir: string, workspacePath: string): Promise { + if (!held) return const now = new Date() - void fs - .utimes(path.join(lockdir, "heartbeat"), now, now) - .catch((err) => log.error("writer lock heartbeat failed", { err })) + void (async () => { + // Stop beating once displaced: without the ownership check a taken-over + // owner keeps refreshing the new owner's heartbeat and may resume + // writing while it believes the lock is still its own. + const current = await read(path.join(lockdir, "meta.json")) + if (!current || current.token !== token) { + held = false + clearInterval(timer) + return + } + await fs.utimes(path.join(lockdir, "heartbeat"), now, now) + })().catch((err) => log.error("writer lock heartbeat failed", { err })) }, BEAT_MS) timer.unref?.() return { release: async () => { + held = false clearInterval(timer) const current = await read(path.join(lockdir, "meta.json")) - if (current && current.token !== token) { - log.error("refusing to release a writer lock owned by another session", { workspacePath }) + if (!current || current.token !== token) { + // The directory (or its meta) belongs to another session now: the + // takeover rewrote it, or it vanished and was reclaimed. Either way + // deleting it would remove a lock this session no longer owns. + if (current) log.error("refusing to release a writer lock owned by another session", { workspacePath }) return } await fs.rm(lockdir, { recursive: true, force: true }) @@ -143,3 +180,14 @@ export async function acquire(dir: string, workspacePath: string): Promise { + const stale = `${lockdir}.stale.${process.pid}.${Date.now()}` + try { + await fs.rename(lockdir, stale) + } catch (err) { + return false + } + void fs.rm(stale, { recursive: true, force: true }).catch((err) => log.error("writer lock stale cleanup failed", { err })) + return claim(lockdir) +} diff --git a/packages/kilo-indexing/test/kilocode/indexing/orchestrator.test.ts b/packages/kilo-indexing/test/kilocode/indexing/orchestrator.test.ts index 59c6da385829..9e838c73421d 100644 --- a/packages/kilo-indexing/test/kilocode/indexing/orchestrator.test.ts +++ b/packages/kilo-indexing/test/kilocode/indexing/orchestrator.test.ts @@ -372,6 +372,7 @@ describe("CodeIndexOrchestrator telemetry", () => { const held = await acquire(dir, "/tmp/ws") expect(held).toBeDefined() + const events: IndexingTelemetryEvent[] = [] const store = new Store("partial", false) let scanned = false const scanner = new Scanner(1, 1, 1) as unknown as DirectoryScanner @@ -388,6 +389,7 @@ describe("CodeIndexOrchestrator telemetry", () => { store as unknown as IVectorStore, scanner, new Watcher() as unknown as IFileWatcher, + (event) => events.push(event), ) await orchestrator.startIndexing("background") @@ -397,6 +399,44 @@ describe("CodeIndexOrchestrator telemetry", () => { expect(store.deleteCount).toBe(0) expect(store.incompleteCount).toBe(0) expect(orchestrator.state).toBe("Standby") + // Contention is observable (not a silent park) and never consumes the + // manager's failure-retry budget: it reports the contention text while a + // real scan error reports its own message. + const error = events.find( + (event): event is Extract => event.type === "error", + ) + expect(error?.error).toContain("Another session is indexing this workspace") + + await held?.release() + await rm(dir, { recursive: true, force: true }) + }) + + test("contended clear reports an error instead of silent Standby", async () => { + const dir = await mkdtemp(path.join(tmpdir(), "indexing-lock-")) + const held = await acquire(dir, "/tmp/ws") + expect(held).toBeDefined() + + const events: IndexingTelemetryEvent[] = [] + const store = new Store("complete", false) + const orchestrator = new CodeIndexOrchestrator( + createConfig(), + new CodeIndexStateManager(), + "/tmp/ws", + { dir, async clearCacheFile() {}, async flush() {} } as unknown as CacheManager, + store as unknown as IVectorStore, + new Scanner(1, 1, 1) as unknown as DirectoryScanner, + new Watcher() as unknown as IFileWatcher, + (event) => events.push(event), + ) + + await orchestrator.clearIndexData() + + expect(store.deleteCount).toBe(0) + expect(orchestrator.state).toBe("Error") + const error = events.find( + (event): event is Extract => event.type === "error", + ) + expect(error?.error).toContain("was not cleared") await held?.release() await rm(dir, { recursive: true, force: true }) diff --git a/packages/kilo-indexing/test/kilocode/indexing/vector-store/qdrant-client.test.ts b/packages/kilo-indexing/test/kilocode/indexing/vector-store/qdrant-client.test.ts index 6661297cb429..c5747c9bbee5 100644 --- a/packages/kilo-indexing/test/kilocode/indexing/vector-store/qdrant-client.test.ts +++ b/packages/kilo-indexing/test/kilocode/indexing/vector-store/qdrant-client.test.ts @@ -831,6 +831,26 @@ describe("QdrantVectorStore", () => { expect(mockCreatePayloadIndex).toHaveBeenCalledTimes(6) }) + test("state reports empty when only the completion marker remains", async () => { + // A scanned-then-cleared collection still holds points_count 1 for the + // completion marker: emptiness must subtract it. + mockGetCollection.mockResolvedValue({ points_count: 1 }) + mockRetrieve.mockResolvedValue([{ payload: { indexing_complete: true } }]) + + const state = await vectorStore.state() + + expect(state).toEqual({ status: "empty", points: 0 }) + }) + + test("state counts data points without the completion marker", async () => { + mockGetCollection.mockResolvedValue({ points_count: 4 }) + mockRetrieve.mockResolvedValue([{ payload: { indexing_complete: true } }]) + + const state = await vectorStore.state() + + expect(state).toEqual({ status: "complete", points: 3 }) + }) + test("should throw vectorDimensionMismatch error when deleteCollection fails during recreation", async () => { const differentVectorSize = 768 mockGetCollection.mockResolvedValue({ diff --git a/packages/kilo-indexing/test/kilocode/indexing/writer-lock.test.ts b/packages/kilo-indexing/test/kilocode/indexing/writer-lock.test.ts index 7e27716595d9..c89bb4a7aeed 100644 --- a/packages/kilo-indexing/test/kilocode/indexing/writer-lock.test.ts +++ b/packages/kilo-indexing/test/kilocode/indexing/writer-lock.test.ts @@ -1,5 +1,5 @@ import { afterEach, beforeEach, describe, expect, test } from "bun:test" -import { mkdir, mkdtemp, readFile, rm, utimes, writeFile } from "fs/promises" +import { mkdir, mkdtemp, readFile, readdir, rm, utimes, writeFile } from "fs/promises" import { hostname, tmpdir } from "os" import path from "path" import { createHash } from "crypto" @@ -93,4 +93,30 @@ describe("writer lock", () => { expect((await meta()).token).toBe(token) await second?.release() }) + + test("reclaims a vanished lock directory instead of wedging", async () => { + const first = await acquire(dir, WS) + expect(first).toBeDefined() + // Winner reclaimed the stale directory out from under a racing claim: + // the loser must fall through to claiming, not report held-forever. + await rm(slot(), { recursive: true, force: true }) + const second = await acquire(dir, WS) + expect(second).toBeDefined() + await second?.release() + }) + + test("a stale release never deletes the current lock directory", async () => { + const first = await acquire(dir, WS) + expect(first).toBeDefined() + // The first owner's directory vanished (taken over); a second session now + // owns the slot. The first owner's release must leave it alone. + await rm(slot(), { recursive: true, force: true }) + const second = await acquire(dir, WS) + const token = (await meta()).token + + await first?.release() + expect((await meta()).token).toBe(token) + expect((await readdir(slot())).sort()).toEqual(["heartbeat", "meta.json"]) + await second?.release() + }) }) From af61ae351e6367ca9a519f60fff286a60798999e Mon Sep 17 00:00:00 2001 From: matt wilkie Date: Fri, 16 Oct 2026 07:05:14 -0700 Subject: [PATCH 05/14] chore(changeset): add trailing newline --- .changeset/single-writer-codebase-index.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.changeset/single-writer-codebase-index.md b/.changeset/single-writer-codebase-index.md index 7cd811e940b4..514f20d71331 100644 --- a/.changeset/single-writer-codebase-index.md +++ b/.changeset/single-writer-codebase-index.md @@ -2,4 +2,4 @@ "@kilocode/kilo-indexing": minor --- -Keep codebase indexing intact when several sessions share a workspace. Only one session now writes a workspace's index, and a session that finds a partially built index no longer discards it and starts over. \ No newline at end of file +Keep codebase indexing intact when several sessions share a workspace. Only one session now writes a workspace's index, and a session that finds a partially built index no longer discards it and starts over. From c5709ad768e56958e8a7acebb41cda3040838492 Mon Sep 17 00:00:00 2001 From: matt wilkie Date: Sat, 17 Oct 2026 07:54:38 -0700 Subject: [PATCH 06/14] fix(indexing): hold writer lock for session lifetime, drain close races --- .../src/indexing/orchestrator.ts | 76 +++++++++++------ .../src/indexing/processors/file-watcher.ts | 6 +- .../indexing/vector-store/qdrant-client.ts | 5 +- .../kilo-indexing/src/indexing/writer-lock.ts | 6 +- .../kilocode/indexing/orchestrator.test.ts | 84 +++++++++++++++++++ 5 files changed, 147 insertions(+), 30 deletions(-) diff --git a/packages/kilo-indexing/src/indexing/orchestrator.ts b/packages/kilo-indexing/src/indexing/orchestrator.ts index d4572bb377f2..462773269359 100644 --- a/packages/kilo-indexing/src/indexing/orchestrator.ts +++ b/packages/kilo-indexing/src/indexing/orchestrator.ts @@ -30,6 +30,11 @@ export class CodeIndexOrchestrator { private _isProcessing = false private _cancelRequested = false private _active?: Promise + // The session's writer lock, held from the first successful run until + // shutdown (or release on error/clear). The file watcher keeps writing the + // store after every scan, so releasing per-run would let a second session + // acquire the lock and scan concurrently — the corruption the lock targets. + private _writer?: Writer constructor( private readonly configManager: CodeIndexConfigManager, @@ -153,9 +158,11 @@ export class CodeIndexOrchestrator { * Only one session may write a workspace's index. Without this, a second * session starts a scan against a store the first one is still filling, * clears it as unusable, and leaves both sessions committing to the same - * table. + * table. The lock is session-held (see _writer): once acquired, later runs + * reuse it instead of contending with themselves. */ private async _acquire(): Promise { + if (this._writer) return this._writer const dir = this.cacheManager.dir if (!dir) { log.warn("no cache directory for the writer lock, proceeding unlocked", { @@ -164,11 +171,20 @@ export class CodeIndexOrchestrator { return { release: async () => {} } } const writer = await acquire(dir, this.workspacePath) - if (writer) return writer + if (writer) { + this._writer = writer + return writer + } log.info("writer lock is held elsewhere", { workspacePath: this.workspacePath }) return undefined } + private async _releaseWriter(): Promise { + const writer = this._writer + this._writer = undefined + await writer?.release() + } + private async runIndexing(trigger: IndexingTelemetryTrigger): Promise { log.info("indexing start requested", { workspacePath: this.workspacePath, @@ -207,13 +223,14 @@ export class CodeIndexOrchestrator { let started = false let source: IndexingTelemetrySource = "watcher" let mode: IndexingTelemetryMode | undefined - let writer: Writer | undefined try { - // Acquired inside the try: a throw between acquire and the body (e.g. - // from state setup below) would otherwise skip the release and keep the - // lock fresh for STALE_MS, blocking every other session. - writer = await this._acquire() + // Acquired inside the try and session-held (see _writer): a throw between + // acquire and the body (e.g. from state setup below) releases below and + // must not leave the lock fresh for STALE_MS, blocking other sessions. + // The lock is NOT released on success: the file watcher keeps writing the + // store after the scan, so it stays held until shutdown or clearIndexData. + const writer = await this._acquire() if (!writer) { // Contention is worth observing, not a silent park: the manager // surfaces start events as activity, so emit an error event the manager @@ -313,9 +330,11 @@ export class CodeIndexOrchestrator { const msg = err instanceof Error ? err.message : "Unknown error" this.stateManager.setSystemState("Error", `Failed during initial scan: ${msg}`) this.stopWatcher() + // The watcher is stopped, so this session no longer writes the store: + // hand the lock back instead of holding it while wedged in Error. + await this._releaseWriter() } finally { this._isProcessing = false - await writer?.release() log.info("indexing start flow finished", { workspacePath: this.workspacePath, state: this.stateManager.state, @@ -408,10 +427,10 @@ export class CodeIndexOrchestrator { this.overlay?.reconcile(this.cacheManager.getAllHashes()) await this.cacheManager.flush?.() await this.vectorStore.markIndexingComplete() - // The scan finished, but this session's file watcher still writes the - // store for every file change. Releasing the writer lock here would let a - // second session acquire it and scan concurrently — the corruption the - // lock targets — so the lock stays held until shutdown or clearIndexData. + // The scan finished, but this session's file watcher keeps writing the + // store for every file change, so the session-held writer lock stays + // acquired until shutdown or clearIndexData (see _writer). Releasing here + // would let a second session acquire it and scan concurrently. this.fileWatcher.setCollecting(true) this.stateManager.setSystemState("Indexed", "File watcher started. Index up-to-date.") log.info("workspace scan finalized", { @@ -438,17 +457,20 @@ export class CodeIndexOrchestrator { this._cancelRequested = true this.scanner.cancel() // Stop the session's own watcher writes before releasing the writer lock: - // _active's release happens in runIndexing's finally, so the store must be - // quiet by then rather than still receiving file-watcher batches. + // the lock releases below, so the store must be quiet by then rather than + // still receiving file-watcher batches. if (this.fileWatcher.shutdown) await this.fileWatcher.shutdown() else this.fileWatcher.dispose() this.fileWatcher.setCollecting(false) - // Close the store before the writer lock releases in _active's finally: - // _active runs runIndexing, whose finally releases the lock, so closing - // here hands the next session a closed connection rather than one whose - // close is still in flight. - await this.vectorStore.close?.() + // Wait out the in-flight scan before closing the store: a scan that is + // still running would lazily reconnect on its next getDb()/getTable() call, + // reopening a connection close() just tore down and leaking it. Closing + // after _active settles keeps the guarantee without racing the scan. The + // next session cannot interleave here: it can only acquire the writer lock + // once _releaseWriter runs below. await this._active + await this.vectorStore.close?.() + await this._releaseWriter() for (const sub of this._fileWatcherSubscriptions) sub.dispose() this._fileWatcherSubscriptions = [] this._isProcessing = false @@ -482,13 +504,19 @@ export class CodeIndexOrchestrator { } public async clearIndexData(): Promise { - const writer = await this._acquire() + // Reuse the session-held lock when this session owns it: acquiring again + // would contend with ourselves. A fresh lock is released below; the + // session-held one stays acquired for the watcher's ongoing writes. + const owned = this._writer !== undefined + const writer = this._writer ?? (await this._acquire()) if (!writer) { // A user-requested clear must not vanish silently when another session - // holds the lock: record contention as an error event so the manager can - // retry or display it, instead of reporting Standby success. + // holds the lock: surface contention as an error event with a + // clear-specific location. The manager only auto-retries startIndexing + // locations, so this stays terminal: the user sees the failure and can + // retry the clear once the other session finishes. this.emitError( - "orchestrator:startIndexing", + "orchestrator:clearIndexData", new Error("Another session is indexing this workspace; index data was not cleared."), "scan", ) @@ -520,7 +548,7 @@ export class CodeIndexOrchestrator { } } finally { this._isProcessing = false - await writer.release() + if (!owned) await this._releaseWriter() log.info("finished clearing index data", { workspacePath: this.workspacePath, state: this.stateManager.state, diff --git a/packages/kilo-indexing/src/indexing/processors/file-watcher.ts b/packages/kilo-indexing/src/indexing/processors/file-watcher.ts index bc344745e349..d07074534b75 100644 --- a/packages/kilo-indexing/src/indexing/processors/file-watcher.ts +++ b/packages/kilo-indexing/src/indexing/processors/file-watcher.ts @@ -317,8 +317,10 @@ export class FileWatcher implements IFileWatcher { */ async shutdown(): Promise { this.collecting = false - if (this.batchProcessDebounceTimer) clearTimeout(this.batchProcessDebounceTimer) - this.batchProcessDebounceTimer = undefined + if (this.batchProcessDebounceTimer) { + clearTimeout(this.batchProcessDebounceTimer) + this.batchProcessDebounceTimer = undefined + } const subscription = this.subscription this.subscription = undefined await subscription diff --git a/packages/kilo-indexing/src/indexing/vector-store/qdrant-client.ts b/packages/kilo-indexing/src/indexing/vector-store/qdrant-client.ts index 581827ea2e51..99967fe5bb55 100644 --- a/packages/kilo-indexing/src/indexing/vector-store/qdrant-client.ts +++ b/packages/kilo-indexing/src/indexing/vector-store/qdrant-client.ts @@ -683,8 +683,8 @@ export class QdrantVectorStore implements IVectorStore { // The completion marker is itself a point, so subtract it before judging // emptiness: otherwise a scanned-then-cleared collection reports // points_count 1 and `empty` is unreachable after any scan. - const marker = (await this.getMetadataPayload()) ? 1 : 0 - const indexed = Math.max(0, points - marker) + const payload = await this.getMetadataPayload() + const indexed = Math.max(0, points - (payload ? 1 : 0)) if (indexed === 0) { log.info("Qdrant collection has no indexed data", { collection: this.collectionName, @@ -693,7 +693,6 @@ export class QdrantVectorStore implements IVectorStore { return { status: "empty", points: indexed } } - const payload = await this.getMetadataPayload() if (payload) { const status = payload[KEY.complete] === true ? "complete" : "partial" log.info("Qdrant indexing metadata evaluated", { diff --git a/packages/kilo-indexing/src/indexing/writer-lock.ts b/packages/kilo-indexing/src/indexing/writer-lock.ts index 3889541958ed..92e2df1e3f4c 100644 --- a/packages/kilo-indexing/src/indexing/writer-lock.ts +++ b/packages/kilo-indexing/src/indexing/writer-lock.ts @@ -186,7 +186,11 @@ async function reclaim(lockdir: string): Promise { try { await fs.rename(lockdir, stale) } catch (err) { - return false + // ENOENT means the directory vanished between the reclaimable check and + // the rename — i.e. the lock is free, not lost to a rival. Claim it + // instead of reporting a lost race and wedging the session in Standby. + if (code(err) !== "ENOENT") return false + return claim(lockdir) } void fs.rm(stale, { recursive: true, force: true }).catch((err) => log.error("writer lock stale cleanup failed", { err })) return claim(lockdir) diff --git a/packages/kilo-indexing/test/kilocode/indexing/orchestrator.test.ts b/packages/kilo-indexing/test/kilocode/indexing/orchestrator.test.ts index 9e838c73421d..257858f34dd7 100644 --- a/packages/kilo-indexing/test/kilocode/indexing/orchestrator.test.ts +++ b/packages/kilo-indexing/test/kilocode/indexing/orchestrator.test.ts @@ -503,4 +503,88 @@ describe("CodeIndexOrchestrator telemetry", () => { expect(store.deleteCount).toBe(0) expect(orchestrator.state).toBe("Error") }) + + test("holds the writer lock across runs until shutdown", async () => { + const dir = await mkdtemp(path.join(tmpdir(), "indexing-lock-")) + const store = new Store("empty", false) + const orchestrator = new CodeIndexOrchestrator( + createConfig(), + new CodeIndexStateManager(), + "/tmp/ws", + { dir, async clearCacheFile() {}, async flush() {} } as unknown as CacheManager, + store as unknown as IVectorStore, + new Scanner(1, 1, 1) as unknown as DirectoryScanner, + new Watcher() as unknown as IFileWatcher, + ) + + await orchestrator.startIndexing("background") + expect(orchestrator.state).toBe("Indexed") + // The lock stays held after the scan for the watcher's ongoing writes: a + // rival session must still see contention instead of scanning concurrently. + expect(await acquire(dir, "/tmp/ws")).toBeUndefined() + + // A second run reuses the session-held lock instead of contending with itself. + await orchestrator.startIndexing("background") + expect(orchestrator.state).toBe("Indexed") + + await orchestrator.shutdown() + // Shutdown hands the lock back: a new session can now acquire it. + const next = await acquire(dir, "/tmp/ws") + expect(next).toBeDefined() + await next?.release() + await rm(dir, { recursive: true, force: true }) + }) + + test("releases the writer lock when a run fails", async () => { + const dir = await mkdtemp(path.join(tmpdir(), "indexing-lock-")) + const store = new Store("empty", false) + const orchestrator = new CodeIndexOrchestrator( + createConfig(), + new CodeIndexStateManager(), + "/tmp/ws", + { dir, async clearCacheFile() {}, async flush() {} } as unknown as CacheManager, + store as unknown as IVectorStore, + new FailScanner() as unknown as DirectoryScanner, + new Watcher() as unknown as IFileWatcher, + ) + + await orchestrator.startIndexing("background") + expect(orchestrator.state).toBe("Error") + // The watcher is stopped on error, so the lock must not stay held while wedged. + const next = await acquire(dir, "/tmp/ws") + expect(next).toBeDefined() + await next?.release() + await rm(dir, { recursive: true, force: true }) + }) + + test("contended clear reports a clear location that the manager does not retry", async () => { + const dir = await mkdtemp(path.join(tmpdir(), "indexing-lock-")) + const held = await acquire(dir, "/tmp/ws") + expect(held).toBeDefined() + + const events: IndexingTelemetryEvent[] = [] + const store = new Store("complete", false) + const orchestrator = new CodeIndexOrchestrator( + createConfig(), + new CodeIndexStateManager(), + "/tmp/ws", + { dir, async clearCacheFile() {}, async flush() {} } as unknown as CacheManager, + store as unknown as IVectorStore, + new Scanner(1, 1, 1) as unknown as DirectoryScanner, + new Watcher() as unknown as IFileWatcher, + (event) => events.push(event), + ) + + await orchestrator.clearIndexData() + + const error = events.find( + (event): event is Extract => event.type === "error", + ) + // A clear-specific location keeps the manager's startIndexing auto-recovery + // from firing: recovery would rerun a scan, not the requested clear. + expect(error?.location).toBe("orchestrator:clearIndexData") + + await held?.release() + await rm(dir, { recursive: true, force: true }) + }) }) From 7be0b6ead97a226d76478c5dd012eb2f6b67ae72 Mon Sep 17 00:00:00 2001 From: maphew <486200+maphew@users.noreply.github.com> Date: Wed, 21 Oct 2026 05:08:53 +0000 Subject: [PATCH 07/14] chore(changeset): add trailing newline --- .changeset/single-writer-codebase-index.md | 1 + 1 file changed, 1 insertion(+) diff --git a/.changeset/single-writer-codebase-index.md b/.changeset/single-writer-codebase-index.md index 514f20d71331..e8a4bd5db9ab 100644 --- a/.changeset/single-writer-codebase-index.md +++ b/.changeset/single-writer-codebase-index.md @@ -3,3 +3,4 @@ --- Keep codebase indexing intact when several sessions share a workspace. Only one session now writes a workspace's index, and a session that finds a partially built index no longer discards it and starts over. + From 8e832b620870b808f280d69575fccbccc3eba189 Mon Sep 17 00:00:00 2001 From: matt wilkie Date: Thu, 22 Oct 2026 13:28:47 -0700 Subject: [PATCH 08/14] fix(indexing): verify cached writer ownership and release lock on clear --- .../src/indexing/orchestrator.ts | 25 +++++--- .../kilo-indexing/src/indexing/writer-lock.ts | 10 +++ .../test/kilocode/indexing/manager.test.ts | 2 +- .../kilocode/indexing/orchestrator.test.ts | 61 ++++++++++++++++++- .../kilocode/indexing/writer-lock.test.ts | 12 ++++ 5 files changed, 100 insertions(+), 10 deletions(-) diff --git a/packages/kilo-indexing/src/indexing/orchestrator.ts b/packages/kilo-indexing/src/indexing/orchestrator.ts index 462773269359..cea5e322da74 100644 --- a/packages/kilo-indexing/src/indexing/orchestrator.ts +++ b/packages/kilo-indexing/src/indexing/orchestrator.ts @@ -162,13 +162,23 @@ export class CodeIndexOrchestrator { * reuse it instead of contending with themselves. */ private async _acquire(): Promise { - if (this._writer) return this._writer + if (this._writer) { + // The session caches its writer across runs. Verify ownership before + // reusing it: a heartbeat that saw another session's token stops beating, + // but only a fresh check here stops this orchestrator from writing the + // store after it has lost the lock. + if (await this._writer.owned()) return this._writer + log.warn("lost the writer lock since the last run; reacquiring", { + workspacePath: this.workspacePath, + }) + this._writer = undefined + } const dir = this.cacheManager.dir if (!dir) { log.warn("no cache directory for the writer lock, proceeding unlocked", { workspacePath: this.workspacePath, }) - return { release: async () => {} } + return { release: async () => {}, owned: async () => true } } const writer = await acquire(dir, this.workspacePath) if (writer) { @@ -505,10 +515,9 @@ export class CodeIndexOrchestrator { public async clearIndexData(): Promise { // Reuse the session-held lock when this session owns it: acquiring again - // would contend with ourselves. A fresh lock is released below; the - // session-held one stays acquired for the watcher's ongoing writes. - const owned = this._writer !== undefined - const writer = this._writer ?? (await this._acquire()) + // would contend with ourselves. Whatever the source, the watcher is + // stopped below, so no writes remain and the lock is released afterwards. + const writer = await this._acquire() if (!writer) { // A user-requested clear must not vanish silently when another session // holds the lock: surface contention as an error event with a @@ -548,7 +557,9 @@ export class CodeIndexOrchestrator { } } finally { this._isProcessing = false - if (!owned) await this._releaseWriter() + // stopWatcher() above means this session no longer writes the store, so + // a session-held lock must be handed back rather than kept until exit. + await this._releaseWriter() log.info("finished clearing index data", { workspacePath: this.workspacePath, state: this.stateManager.state, diff --git a/packages/kilo-indexing/src/indexing/writer-lock.ts b/packages/kilo-indexing/src/indexing/writer-lock.ts index 92e2df1e3f4c..edb6cca1d4b9 100644 --- a/packages/kilo-indexing/src/indexing/writer-lock.ts +++ b/packages/kilo-indexing/src/indexing/writer-lock.ts @@ -11,6 +11,7 @@ const BEAT_MS = Math.floor(STALE_MS / 3) export type Writer = { release: () => Promise + owned: () => Promise } type Meta = { @@ -163,7 +164,16 @@ export async function acquire(dir: string, workspacePath: string): Promise { + const current = await read(path.join(lockdir, "meta.json")) + return current?.token === token + } + return { + owned, release: async () => { held = false clearInterval(timer) diff --git a/packages/kilo-indexing/test/kilocode/indexing/manager.test.ts b/packages/kilo-indexing/test/kilocode/indexing/manager.test.ts index 649d4a1444ea..63209cf365d5 100644 --- a/packages/kilo-indexing/test/kilocode/indexing/manager.test.ts +++ b/packages/kilo-indexing/test/kilocode/indexing/manager.test.ts @@ -210,7 +210,7 @@ describe("CodeIndexManager", () => { const timeout = setTimeout(() => { child.kill() gate.reject(new Error("Indexing subprocess timed out")) - }, 10_000) + }, 30_000) const [exit, stderr] = await Promise.race([ Promise.all([child.exited, new Response(child.stderr).text()]), gate.promise, diff --git a/packages/kilo-indexing/test/kilocode/indexing/orchestrator.test.ts b/packages/kilo-indexing/test/kilocode/indexing/orchestrator.test.ts index 257858f34dd7..2a1e4a16030d 100644 --- a/packages/kilo-indexing/test/kilocode/indexing/orchestrator.test.ts +++ b/packages/kilo-indexing/test/kilocode/indexing/orchestrator.test.ts @@ -1,6 +1,7 @@ import { describe, expect, test } from "bun:test" -import { mkdtemp, rm } from "fs/promises" -import { tmpdir } from "os" +import { mkdir, mkdtemp, rm, writeFile } from "fs/promises" +import { createHash } from "crypto" +import { hostname, tmpdir } from "os" import path from "path" import { CodeIndexConfigManager } from "../../../src/indexing/config-manager" import { CodeIndexOrchestrator } from "../../../src/indexing/orchestrator" @@ -535,6 +536,62 @@ describe("CodeIndexOrchestrator telemetry", () => { await rm(dir, { recursive: true, force: true }) }) + test("releases the session-held writer lock after clear", async () => { + const dir = await mkdtemp(path.join(tmpdir(), "indexing-lock-")) + const store = new Store("empty", false) + const orchestrator = new CodeIndexOrchestrator( + createConfig(), + new CodeIndexStateManager(), + "/tmp/ws", + { dir, async clearCacheFile() {}, async flush() {} } as unknown as CacheManager, + store as unknown as IVectorStore, + new Scanner(1, 1, 1) as unknown as DirectoryScanner, + new Watcher() as unknown as IFileWatcher, + ) + + await orchestrator.startIndexing("background") + expect(orchestrator.state).toBe("Indexed") + + await orchestrator.clearIndexData() + // The watcher is stopped by clear, so no writes remain: keeping the lock + // would starve every other session for the rest of this one's lifetime. + const next = await acquire(dir, "/tmp/ws") + expect(next).toBeDefined() + await next?.release() + await rm(dir, { recursive: true, force: true }) + }) + + test("stops writing when another session takes over the cached lock", async () => { + const dir = await mkdtemp(path.join(tmpdir(), "indexing-lock-")) + const store = new Store("complete", false) + const orchestrator = new CodeIndexOrchestrator( + createConfig(), + new CodeIndexStateManager(), + "/tmp/ws", + { dir, async clearCacheFile() {}, async flush() {} } as unknown as CacheManager, + store as unknown as IVectorStore, + new Scanner(1, 1, 1) as unknown as DirectoryScanner, + new Watcher() as unknown as IFileWatcher, + ) + + await orchestrator.startIndexing("background") + expect(orchestrator.state).toBe("Indexed") + + const slot = path.join(dir, "locks", createHash("sha256").update("/tmp/ws").digest("hex").substring(0, 16)) + const before = store.deleteCount + await rm(slot, { recursive: true, force: true }) + await mkdir(slot, { recursive: true }) + await writeFile(path.join(slot, "meta.json"), JSON.stringify({ token: "rival", pid: process.pid, host: hostname() })) + await writeFile(path.join(slot, "heartbeat"), "") + + await orchestrator.clearIndexData() + // The cached writer is no longer the owner, so the clear must not write + // the store. It re-acquires and reports contention instead. + expect(store.deleteCount).toBe(before) + expect(orchestrator.state).toBe("Error") + await rm(dir, { recursive: true, force: true }) + }) + test("releases the writer lock when a run fails", async () => { const dir = await mkdtemp(path.join(tmpdir(), "indexing-lock-")) const store = new Store("empty", false) diff --git a/packages/kilo-indexing/test/kilocode/indexing/writer-lock.test.ts b/packages/kilo-indexing/test/kilocode/indexing/writer-lock.test.ts index c89bb4a7aeed..a28e51fe6be8 100644 --- a/packages/kilo-indexing/test/kilocode/indexing/writer-lock.test.ts +++ b/packages/kilo-indexing/test/kilocode/indexing/writer-lock.test.ts @@ -60,6 +60,18 @@ describe("writer lock", () => { await writer?.release() }) + test("reports ownership only while the lock still belongs to this writer", async () => { + const writer = await acquire(dir, WS) + expect(await writer?.owned()).toBe(true) + + // A takeover rewrites the metadata; the cached writer must notice instead + // of continuing to write the store as if the lock were still its own. + const current = await meta() + await writeFile(path.join(slot(), "meta.json"), JSON.stringify({ ...current, token: "rival" })) + expect(await writer?.owned()).toBe(false) + await writer?.release() + }) + test("takes over a lock whose heartbeat stopped", async () => { await plant({ token: "dead", pid: process.pid, host: "elsewhere" }, true) From 014a2b0becbd5b4bc0f8cd2bd0446e68edecb42c Mon Sep 17 00:00:00 2001 From: matt wilkie Date: Thu, 22 Oct 2026 13:47:38 -0700 Subject: [PATCH 09/14] test(indexing): exit the keyless probe after dispose on Windows --- .../kilo-indexing/test/kilocode/indexing/manager.test.ts | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/packages/kilo-indexing/test/kilocode/indexing/manager.test.ts b/packages/kilo-indexing/test/kilocode/indexing/manager.test.ts index 63209cf365d5..9c64a83c686b 100644 --- a/packages/kilo-indexing/test/kilocode/indexing/manager.test.ts +++ b/packages/kilo-indexing/test/kilocode/indexing/manager.test.ts @@ -199,6 +199,11 @@ describe("CodeIndexManager", () => { } finally { await mgr.dispose() } + // The manager can leave a native watcher handle alive on Windows even + // after dispose. This is a one-shot probe, so exit explicitly instead + // of waiting for the event loop to drain; otherwise the parent sees a + // live child and times out while all assertions have already passed. + process.exit(0) ` const child = Bun.spawn([process.execPath, "-e", script], { cwd: join(import.meta.dir, "../../.."), From 715dd3bb64927a9bb4b2dd8429b91eef0c8e178a Mon Sep 17 00:00:00 2001 From: matt wilkie Date: Thu, 22 Oct 2026 14:38:28 -0700 Subject: [PATCH 10/14] fix(indexing): verify lock ownership per watcher batch and drain scans before clear --- .../src/indexing/orchestrator.ts | 33 ++++++++ .../kilocode/indexing/orchestrator.test.ts | 84 +++++++++++++++++++ 2 files changed, 117 insertions(+) diff --git a/packages/kilo-indexing/src/indexing/orchestrator.ts b/packages/kilo-indexing/src/indexing/orchestrator.ts index cea5e322da74..65b004b58b6a 100644 --- a/packages/kilo-indexing/src/indexing/orchestrator.ts +++ b/packages/kilo-indexing/src/indexing/orchestrator.ts @@ -106,6 +106,11 @@ export class CodeIndexOrchestrator { workspacePath: this.workspacePath, filesInBatch: paths.length, }) + // The watcher writes the store on its own after the scan, outside + // runIndexing. Re-check ownership before every batch: a heartbeat can + // lapse (event loop stall) long enough for another session to reclaim + // the lock, and this session must not keep committing without it. + void this._verifyWriter() if (this.stateManager.state !== "Indexing") { this.stateManager.setSystemState("Indexing", "Processing file changes...") } @@ -195,6 +200,27 @@ export class CodeIndexOrchestrator { await writer?.release() } + /** + * Stops the watcher when this session no longer owns the writer lock. + * + * The heartbeat stops beating after a takeover, but the watcher keeps + * receiving file events and writing the store. Checking once per batch + * bounds how much can be written after ownership is lost; shutting the + * watcher down turns a silent split-brain into a visible Error state. + */ + private async _verifyWriter(): Promise { + const writer = this._writer + if (!writer) return + if (await writer.owned()) return + if (this._writer !== writer) return + this._writer = undefined + log.error("writer lock lost; stopping the file watcher to avoid unlocked writes", { + workspacePath: this.workspacePath, + }) + this.stateManager.setSystemState("Error", "Lost the indexing writer lock to another session. File watching stopped.") + this.stopWatcher() + } + private async runIndexing(trigger: IndexingTelemetryTrigger): Promise { log.info("indexing start requested", { workspacePath: this.workspacePath, @@ -537,6 +563,13 @@ export class CodeIndexOrchestrator { log.info("clearing index data", { workspacePath: this.workspacePath }) try { + // A clear must not race an in-flight scan. Cancel it and wait for the + // scan to settle before deleting the collection: otherwise batches still + // running can re-add points after the clear and write under a lock this + // method is about to release (shutdown() waits the same way). + this._cancelRequested = true + this.scanner.cancel() + await this._active this.stopWatcher() try { diff --git a/packages/kilo-indexing/test/kilocode/indexing/orchestrator.test.ts b/packages/kilo-indexing/test/kilocode/indexing/orchestrator.test.ts index 2a1e4a16030d..7616a1e617b1 100644 --- a/packages/kilo-indexing/test/kilocode/indexing/orchestrator.test.ts +++ b/packages/kilo-indexing/test/kilocode/indexing/orchestrator.test.ts @@ -107,6 +107,7 @@ class Scanner { } class Watcher { + public disposed = 0 public readonly onDidStartBatchProcessing = new Emitter() public readonly onBatchProgressUpdate = new Emitter<{ processedInBatch: number @@ -128,6 +129,7 @@ class Watcher { } dispose(): void { + this.disposed += 1 this.onDidStartBatchProcessing.dispose() this.onBatchProgressUpdate.dispose() this.onDidFinishBatchProcessing.dispose() @@ -155,6 +157,28 @@ class BlockingScanner { updateBatchSegmentThreshold(_newThreshold: number): void {} } +class StubbornScanner { + public isCancelled = false + public finished = false + readonly started = Promise.withResolvers() + readonly gate = Promise.withResolvers() + + async scanDirectory(): Promise<{ stats: { processed: number; skipped: number }; totalBlockCount: number }> { + this.started.resolve() + await this.gate.promise + this.finished = true + return { stats: { processed: 0, skipped: 0 }, totalBlockCount: 0 } + } + + cancel(): void { + // A real batch already in flight finishes its current work even after + // cancellation; only the caller releasing the gate ends this scan. + this.isCancelled = true + } + + updateBatchSegmentThreshold(_newThreshold: number): void {} +} + class FailScanner { public readonly isCancelled = false @@ -561,6 +585,66 @@ describe("CodeIndexOrchestrator telemetry", () => { await rm(dir, { recursive: true, force: true }) }) + test("waits for the active scan before clearing data", async () => { + const dir = await mkdtemp(path.join(tmpdir(), "indexing-lock-")) + const scanner = new StubbornScanner() + const store = new Store("complete", false) + const orchestrator = new CodeIndexOrchestrator( + createConfig(), + new CodeIndexStateManager(), + "/tmp/ws", + { dir, async clearCacheFile() {}, async flush() {} } as unknown as CacheManager, + store as unknown as IVectorStore, + scanner as unknown as DirectoryScanner, + new Watcher() as unknown as IFileWatcher, + ) + + const active = orchestrator.startIndexing("background") + await scanner.started.promise + const clearing = orchestrator.clearIndexData() + await Bun.sleep(20) + // The in-flight batch is still running, so nothing may be deleted yet. + expect(store.deleteCount).toBe(0) + + scanner.gate.resolve() + await clearing + await active + expect(store.deleteCount).toBe(1) + await rm(dir, { recursive: true, force: true }) + }) + + test("stops the watcher when the writer lock is lost during a batch", async () => { + const dir = await mkdtemp(path.join(tmpdir(), "indexing-lock-")) + const store = new Store("complete", false) + const watcher = new Watcher() + const orchestrator = new CodeIndexOrchestrator( + createConfig(), + new CodeIndexStateManager(), + "/tmp/ws", + { dir, async clearCacheFile() {}, async flush() {} } as unknown as CacheManager, + store as unknown as IVectorStore, + new Scanner(1, 1, 1) as unknown as DirectoryScanner, + watcher as unknown as IFileWatcher, + ) + + await orchestrator.startIndexing("background") + expect(orchestrator.state).toBe("Indexed") + + const slot = path.join(dir, "locks", createHash("sha256").update("/tmp/ws").digest("hex").substring(0, 16)) + await rm(slot, { recursive: true, force: true }) + await mkdir(slot, { recursive: true }) + await writeFile(path.join(slot, "meta.json"), JSON.stringify({ token: "rival", pid: process.pid, host: hostname() })) + await writeFile(path.join(slot, "heartbeat"), "") + + watcher.onDidStartBatchProcessing.fire(["src/a.ts"]) + await Bun.sleep(20) + // The watcher writes the store outside runIndexing, so a batch after the + // takeover must shut it down rather than commit unlocked. + expect(watcher.disposed).toBeGreaterThan(0) + expect(orchestrator.state).toBe("Error") + await rm(dir, { recursive: true, force: true }) + }) + test("stops writing when another session takes over the cached lock", async () => { const dir = await mkdtemp(path.join(tmpdir(), "indexing-lock-")) const store = new Store("complete", false) From 1342f5817af701e810217ab43f582fb80a2a8bdf Mon Sep 17 00:00:00 2001 From: matt wilkie Date: Thu, 22 Oct 2026 14:56:08 -0700 Subject: [PATCH 11/14] fix(indexing): drain watcher batches before clear and tolerate a blip reading lock metadata --- packages/kilo-indexing/src/indexing/orchestrator.ts | 4 ++++ packages/kilo-indexing/src/indexing/writer-lock.ts | 11 +++++++++-- .../test/kilocode/indexing/orchestrator.test.ts | 2 +- 3 files changed, 14 insertions(+), 3 deletions(-) diff --git a/packages/kilo-indexing/src/indexing/orchestrator.ts b/packages/kilo-indexing/src/indexing/orchestrator.ts index 65b004b58b6a..1fc7279956b4 100644 --- a/packages/kilo-indexing/src/indexing/orchestrator.ts +++ b/packages/kilo-indexing/src/indexing/orchestrator.ts @@ -570,6 +570,10 @@ export class CodeIndexOrchestrator { this._cancelRequested = true this.scanner.cancel() await this._active + // The watcher's drainTask is separate from _active: shutdown() waits for + // it before returning, so a batch already writing cannot upsert after + // deleteCollection() or after the lock is released below. + if (this.fileWatcher.shutdown) await this.fileWatcher.shutdown() this.stopWatcher() try { diff --git a/packages/kilo-indexing/src/indexing/writer-lock.ts b/packages/kilo-indexing/src/indexing/writer-lock.ts index edb6cca1d4b9..329dc8077cef 100644 --- a/packages/kilo-indexing/src/indexing/writer-lock.ts +++ b/packages/kilo-indexing/src/indexing/writer-lock.ts @@ -168,8 +168,15 @@ export async function acquire(dir: string, workspacePath: string): Promise { - const current = await read(path.join(lockdir, "meta.json")) - return current?.token === token + // A single unreadable metadata read (transient IO, a writer mid-rewrite) + // must not permanently park the session and dispose its watcher. Retry + // once; a real takeover leaves a stable foreign token on the second read. + for (let attempt = 0; attempt < 2; attempt += 1) { + const current = await read(path.join(lockdir, "meta.json")) + if (current) return current.token === token + if (attempt === 0) await new Promise((resolve) => setTimeout(resolve, 25)) + } + return false } return { diff --git a/packages/kilo-indexing/test/kilocode/indexing/orchestrator.test.ts b/packages/kilo-indexing/test/kilocode/indexing/orchestrator.test.ts index 7616a1e617b1..9015403541b4 100644 --- a/packages/kilo-indexing/test/kilocode/indexing/orchestrator.test.ts +++ b/packages/kilo-indexing/test/kilocode/indexing/orchestrator.test.ts @@ -637,7 +637,7 @@ describe("CodeIndexOrchestrator telemetry", () => { await writeFile(path.join(slot, "heartbeat"), "") watcher.onDidStartBatchProcessing.fire(["src/a.ts"]) - await Bun.sleep(20) + for (let i = 0; i < 100 && watcher.disposed === 0; i += 1) await Bun.sleep(10) // The watcher writes the store outside runIndexing, so a batch after the // takeover must shut it down rather than commit unlocked. expect(watcher.disposed).toBeGreaterThan(0) From 915e6554d49c487b165759b658a5e696724404ff Mon Sep 17 00:00:00 2001 From: matt wilkie Date: Thu, 22 Oct 2026 15:07:41 -0700 Subject: [PATCH 12/14] fix(indexing): re-acquire the writer lock before clearing after a failed scan --- .../src/indexing/orchestrator.ts | 18 +++++++ .../kilocode/indexing/orchestrator.test.ts | 54 +++++++++++++++++++ 2 files changed, 72 insertions(+) diff --git a/packages/kilo-indexing/src/indexing/orchestrator.ts b/packages/kilo-indexing/src/indexing/orchestrator.ts index 1fc7279956b4..d254ecf10aca 100644 --- a/packages/kilo-indexing/src/indexing/orchestrator.ts +++ b/packages/kilo-indexing/src/indexing/orchestrator.ts @@ -570,6 +570,24 @@ export class CodeIndexOrchestrator { this._cancelRequested = true this.scanner.cancel() await this._active + // The scan's error path releases the writer lock. Re-acquire before + // deleting: otherwise another session can take the lock between that + // release and this clear, and we would delete its collection unlocked. + if (!this._writer) { + const retry = await this._acquire() + if (!retry) { + this.emitError( + "orchestrator:clearIndexData", + new Error("Another session is indexing this workspace; index data was not cleared."), + "scan", + ) + this.stateManager.setSystemState( + "Error", + "Another session is indexing this workspace; index data was not cleared.", + ) + return + } + } // The watcher's drainTask is separate from _active: shutdown() waits for // it before returning, so a batch already writing cannot upsert after // deleteCollection() or after the lock is released below. diff --git a/packages/kilo-indexing/test/kilocode/indexing/orchestrator.test.ts b/packages/kilo-indexing/test/kilocode/indexing/orchestrator.test.ts index 9015403541b4..ac1c561945d5 100644 --- a/packages/kilo-indexing/test/kilocode/indexing/orchestrator.test.ts +++ b/packages/kilo-indexing/test/kilocode/indexing/orchestrator.test.ts @@ -179,6 +179,25 @@ class StubbornScanner { updateBatchSegmentThreshold(_newThreshold: number): void {} } +class FailingBlockingScanner { + public isCancelled = false + readonly started = Promise.withResolvers() + readonly gate = Promise.withResolvers() + + async scanDirectory(): Promise<{ stats: { processed: number; skipped: number }; totalBlockCount: number }> { + this.started.resolve() + await this.gate.promise + throw new Error("scan failed") + } + + cancel(): void { + this.isCancelled = true + this.gate.resolve() + } + + updateBatchSegmentThreshold(_newThreshold: number): void {} +} + class FailScanner { public readonly isCancelled = false @@ -613,6 +632,41 @@ describe("CodeIndexOrchestrator telemetry", () => { await rm(dir, { recursive: true, force: true }) }) + test("re-acquires the writer lock before clearing after a failed scan", async () => { + const dir = await mkdtemp(path.join(tmpdir(), "indexing-lock-")) + const scanner = new FailingBlockingScanner() + const store = new Store("complete", false) + let heldDuringDelete = false + const remove = store.deleteCollection.bind(store) + store.deleteCollection = async () => { + heldDuringDelete = (await acquire(dir, "/tmp/ws")) === undefined + await remove() + } + const orchestrator = new CodeIndexOrchestrator( + createConfig(), + new CodeIndexStateManager(), + "/tmp/ws", + { dir, async clearCacheFile() {}, async flush() {} } as unknown as CacheManager, + store as unknown as IVectorStore, + scanner as unknown as DirectoryScanner, + new Watcher() as unknown as IFileWatcher, + ) + + const active = orchestrator.startIndexing("background") + await scanner.started.promise + // The scan's error path releases the session lock; the clear must take it + // again rather than deleting the collection unlocked. + await orchestrator.clearIndexData() + await active + + expect(store.deleteCount).toBe(1) + expect(heldDuringDelete).toBe(true) + const next = await acquire(dir, "/tmp/ws") + expect(next).toBeDefined() + await next?.release() + await rm(dir, { recursive: true, force: true }) + }) + test("stops the watcher when the writer lock is lost during a batch", async () => { const dir = await mkdtemp(path.join(tmpdir(), "indexing-lock-")) const store = new Store("complete", false) From e00de4c37cd7556b3982575eb8e8aca4cbee36b5 Mon Sep 17 00:00:00 2001 From: matt wilkie Date: Fri, 23 Oct 2026 07:45:22 -0700 Subject: [PATCH 13/14] fix(indexing): verify cached writer ownership before clearing --- .../src/indexing/orchestrator.ts | 33 +++++++++-------- .../kilocode/indexing/orchestrator.test.ts | 36 +++++++++++++++++++ 2 files changed, 52 insertions(+), 17 deletions(-) diff --git a/packages/kilo-indexing/src/indexing/orchestrator.ts b/packages/kilo-indexing/src/indexing/orchestrator.ts index d254ecf10aca..e52fc3958050 100644 --- a/packages/kilo-indexing/src/indexing/orchestrator.ts +++ b/packages/kilo-indexing/src/indexing/orchestrator.ts @@ -570,23 +570,22 @@ export class CodeIndexOrchestrator { this._cancelRequested = true this.scanner.cancel() await this._active - // The scan's error path releases the writer lock. Re-acquire before - // deleting: otherwise another session can take the lock between that - // release and this clear, and we would delete its collection unlocked. - if (!this._writer) { - const retry = await this._acquire() - if (!retry) { - this.emitError( - "orchestrator:clearIndexData", - new Error("Another session is indexing this workspace; index data was not cleared."), - "scan", - ) - this.stateManager.setSystemState( - "Error", - "Another session is indexing this workspace; index data was not cleared.", - ) - return - } + // Verify ownership before deleting even when a writer is still cached: + // a heartbeat lapse during the awaited scan can let another session + // reclaim the lock while _writer stays set. _acquire() re-checks owned() + // and reuses the cached writer only when this session still holds it. + const retry = await this._acquire() + if (!retry) { + this.emitError( + "orchestrator:clearIndexData", + new Error("Another session is indexing this workspace; index data was not cleared."), + "scan", + ) + this.stateManager.setSystemState( + "Error", + "Another session is indexing this workspace; index data was not cleared.", + ) + return } // The watcher's drainTask is separate from _active: shutdown() waits for // it before returning, so a batch already writing cannot upsert after diff --git a/packages/kilo-indexing/test/kilocode/indexing/orchestrator.test.ts b/packages/kilo-indexing/test/kilocode/indexing/orchestrator.test.ts index ac1c561945d5..34347f7cf0cf 100644 --- a/packages/kilo-indexing/test/kilocode/indexing/orchestrator.test.ts +++ b/packages/kilo-indexing/test/kilocode/indexing/orchestrator.test.ts @@ -108,6 +108,8 @@ class Scanner { class Watcher { public disposed = 0 + public shutdowns = 0 + public onShutdown?: () => Promise public readonly onDidStartBatchProcessing = new Emitter() public readonly onBatchProgressUpdate = new Emitter<{ processedInBatch: number @@ -128,6 +130,11 @@ class Watcher { } } + async shutdown(): Promise { + this.shutdowns += 1 + await this.onShutdown?.() + } + dispose(): void { this.disposed += 1 this.onDidStartBatchProcessing.dispose() @@ -667,6 +674,35 @@ describe("CodeIndexOrchestrator telemetry", () => { await rm(dir, { recursive: true, force: true }) }) + test("waits for the watcher drain before clearing data", async () => { + const dir = await mkdtemp(path.join(tmpdir(), "indexing-lock-")) + const store = new Store("complete", false) + const watcher = new Watcher() + const drain = Promise.withResolvers() + watcher.onShutdown = () => drain.promise + const orchestrator = new CodeIndexOrchestrator( + createConfig(), + new CodeIndexStateManager(), + "/tmp/ws", + { dir, async clearCacheFile() {}, async flush() {} } as unknown as CacheManager, + store as unknown as IVectorStore, + new Scanner(1, 1, 1) as unknown as DirectoryScanner, + watcher as unknown as IFileWatcher, + ) + + await orchestrator.startIndexing("background") + const clearing = orchestrator.clearIndexData() + await Bun.sleep(20) + // The watcher's drain task is still writing, so the collection stays. + expect(watcher.shutdowns).toBe(1) + expect(store.deleteCount).toBe(0) + + drain.resolve() + await clearing + expect(store.deleteCount).toBe(1) + await rm(dir, { recursive: true, force: true }) + }) + test("stops the watcher when the writer lock is lost during a batch", async () => { const dir = await mkdtemp(path.join(tmpdir(), "indexing-lock-")) const store = new Store("complete", false) From 58be06810ded170cb104a8bcf0ceadf34b29dc4b Mon Sep 17 00:00:00 2001 From: matt wilkie Date: Sat, 24 Oct 2026 07:03:01 -0700 Subject: [PATCH 14/14] test(indexing): poll for cancel and drain instead of fixed sleeps --- .../test/kilocode/indexing/orchestrator.test.ts | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/packages/kilo-indexing/test/kilocode/indexing/orchestrator.test.ts b/packages/kilo-indexing/test/kilocode/indexing/orchestrator.test.ts index 34347f7cf0cf..a75f2ae476dd 100644 --- a/packages/kilo-indexing/test/kilocode/indexing/orchestrator.test.ts +++ b/packages/kilo-indexing/test/kilocode/indexing/orchestrator.test.ts @@ -628,8 +628,9 @@ describe("CodeIndexOrchestrator telemetry", () => { const active = orchestrator.startIndexing("background") await scanner.started.promise const clearing = orchestrator.clearIndexData() - await Bun.sleep(20) + for (let i = 0; i < 200 && !scanner.isCancelled; i += 1) await Bun.sleep(5) // The in-flight batch is still running, so nothing may be deleted yet. + expect(scanner.isCancelled).toBe(true) expect(store.deleteCount).toBe(0) scanner.gate.resolve() @@ -692,7 +693,7 @@ describe("CodeIndexOrchestrator telemetry", () => { await orchestrator.startIndexing("background") const clearing = orchestrator.clearIndexData() - await Bun.sleep(20) + for (let i = 0; i < 200 && watcher.shutdowns === 0; i += 1) await Bun.sleep(5) // The watcher's drain task is still writing, so the collection stays. expect(watcher.shutdowns).toBe(1) expect(store.deleteCount).toBe(0)