fix(core): serialize task history updates across parallel tabs

Author: easonliang28

src/core/task-persistence/TaskHistoryLock.ts

@@ -0,0 +1,100 @@
+import * as fs from "fs/promises"
+import * as path from "path"
+import * as lockfile from "proper-lockfile"
+
+import { GlobalFileNames } from "../../shared/globalFileNames"
+import { getStorageBasePath } from "../../utils/storage"
+
+/**
+ * Cross-process lock for task history mutations.
+ *
+ * Multiple `ClineProvider` instances may live in separate extension-host
+ * processes (for example VS Code windows) while sharing the same task history
+ * storage. Each process has its own `TaskHistoryStore`, so an in-memory mutex is
+ * not sufficient. This lock serializes mutations by taking an exclusive advisory
+ * lock on the shared `tasks/_history.lock` file.
+ */
+export class TaskHistoryLock {
+	private queue: Promise<unknown> = Promise.resolve()
+
+	/**
+	 * Acquires the shared task-history lock and executes `fn` while holding it.
+	 *
+	 * The lock file is scoped to the effective storage root (including custom
+	 * storage path resolution) so all windows/processes targeting the same history
+	 * store contend on the same file.
+	 */
+	async withLock<T>(globalStoragePath: string, fn: () => Promise<T>): Promise<T> {
+		const result = this.queue.then(
+			async () => {
+				const lockFilePath = await this.getLockFilePath(globalStoragePath)
+				return this.runWithFileLock(lockFilePath, fn)
+			},
+			async () => {
+				const lockFilePath = await this.getLockFilePath(globalStoragePath)
+				return this.runWithFileLock(lockFilePath, fn)
+			},
+		)
+
+		this.queue = result.then(
+			() => undefined,
+			() => undefined,
+		)
+
+		return result
+	}
+
+	/**
+	 * Clears in-process queues. File locks held by other processes are not affected.
+	 */
+	reset(): void {
+		this.queue = Promise.resolve()
+	}
+
+	async getLockFilePath(globalStoragePath: string): Promise<string> {
+		const basePath = await getStorageBasePath(globalStoragePath)
+		const tasksDir = path.join(basePath, "tasks")
+		await fs.mkdir(tasksDir, { recursive: true })
+		const lockFilePath = path.join(tasksDir, GlobalFileNames.historyLock)
+
+		try {
+			await fs.open(lockFilePath, "a").then((handle) => handle.close())
+		} catch (error) {
+			console.error(`[TaskHistoryLock] Failed to create lock file at ${lockFilePath}:`, error)
+			throw error
+		}
+
+		return lockFilePath
+	}
+
+	private async runWithFileLock<T>(lockFilePath: string, fn: () => Promise<T>): Promise<T> {
+		let releaseLock: (() => Promise<void>) | undefined
+
+		try {
+			releaseLock = await lockfile.lock(lockFilePath, {
+				stale: 31000,
+				update: 10000,
+				realpath: false,
+				retries: {
+					retries: 10,
+					factor: 2,
+					minTimeout: 50,
+					maxTimeout: 1000,
+				},
+				onCompromised: (err) => {
+					console.error(`[TaskHistoryLock] Lock at ${lockFilePath} was compromised:`, err)
+					throw err
+				},
+			})
+
+			return await fn()
+		} finally {
+			if (releaseLock) {
+				await releaseLock()
+			}
+		}
+	}
+}
+
+// Singleton instance shared across all ClineProvider instances in this process.
+export const taskHistoryLock = new TaskHistoryLock()

src/core/task-persistence/__tests__/TaskHistoryLock.spec.ts

@@ -0,0 +1,153 @@
+import * as fs from "fs/promises"
+import * as os from "os"
+import * as path from "path"
+import { fork, type ChildProcess } from "child_process"
+
+import { TaskHistoryLock } from "../TaskHistoryLock"
+
+vi.mock("../../../utils/storage", () => ({
+	getStorageBasePath: vi.fn().mockImplementation((defaultPath: string) => defaultPath),
+}))
+
+const waitForMessage = (child: ChildProcess, expected: string): Promise<void> =>
+	new Promise((resolve, reject) => {
+		const timeout = setTimeout(() => {
+			child.off("message", onMessage)
+			reject(new Error(`Timed out waiting for child process message: ${expected}`))
+		}, 5000)
+
+		const onMessage = (message: unknown) => {
+			if (message === expected) {
+				clearTimeout(timeout)
+				child.off("message", onMessage)
+				resolve()
+			}
+		}
+
+		child.on("message", onMessage)
+	})
+
+describe("TaskHistoryLock", () => {
+	let tmpDir: string
+
+	beforeEach(async () => {
+		tmpDir = await fs.mkdtemp(path.join(os.tmpdir(), "task-history-lock-"))
+	})
+
+	afterEach(async () => {
+		await fs.rm(tmpDir, { recursive: true, force: true })
+	})
+
+	it("serializes concurrent operations", async () => {
+		const lock = new TaskHistoryLock()
+		let activeCount = 0
+		let maxActiveCount = 0
+		const order: string[] = []
+
+		const run = (id: string) =>
+			lock.withLock(tmpDir, async () => {
+				activeCount++
+				maxActiveCount = Math.max(maxActiveCount, activeCount)
+				order.push(`start:${id}`)
+				await new Promise((resolve) => setTimeout(resolve, 5))
+				order.push(`end:${id}`)
+				activeCount--
+				return id
+			})
+
+		const results = await Promise.all([run("a"), run("b"), run("c")])
+
+		expect(results).toEqual(["a", "b", "c"])
+		expect(maxActiveCount).toBe(1)
+		expect(order).toHaveLength(6)
+		for (const id of ["a", "b", "c"]) {
+			expect(order).toContain(`start:${id}`)
+			expect(order).toContain(`end:${id}`)
+			expect(order.indexOf(`start:${id}`)).toBeLessThan(order.indexOf(`end:${id}`))
+		}
+	})
+
+	it("continues processing after a previous operation rejects", async () => {
+		const lock = new TaskHistoryLock()
+		const order: string[] = []
+
+		const failed = lock.withLock(tmpDir, async () => {
+			order.push("start:fail")
+			throw new Error("simulated failure")
+		})
+
+		const succeeded = lock.withLock(tmpDir, async () => {
+			order.push("start:success")
+			return "ok"
+		})
+
+		await expect(failed).rejects.toThrow("simulated failure")
+		await expect(succeeded).resolves.toBe("ok")
+		expect(order).toEqual(["start:fail", "start:success"])
+	})
+
+	it("waits for an independent process holding the same lock file", async () => {
+		const lock = new TaskHistoryLock()
+		const lockFilePath = await lock.getLockFilePath(tmpDir)
+		const childScriptPath = path.join(tmpDir, "hold-history-lock.cjs")
+		await fs.writeFile(
+			childScriptPath,
+			`
+const lockfile = require("proper-lockfile")
+
+let release
+
+async function main() {
+	release = await lockfile.lock(process.argv[2], {
+		stale: 31000,
+		update: 10000,
+		realpath: false,
+	})
+	process.send?.("locked")
+}
+
+process.on("message", async (message) => {
+	if (message === "release") {
+		await release?.()
+		process.send?.("released")
+		process.exit(0)
+	}
+})
+
+main().catch((error) => {
+	process.send?.({ error: error instanceof Error ? error.message : String(error) })
+	process.exit(1)
+})
+`,
+			"utf8",
+		)
+
+		const child = fork(childScriptPath, [lockFilePath], { stdio: ["ignore", "ignore", "ignore", "ipc"] })
+		try {
+			await waitForMessage(child, "locked")
+
+			let enteredCriticalSection = false
+			const blocked = lock.withLock(tmpDir, async () => {
+				enteredCriticalSection = true
+				return "acquired"
+			})
+
+			await new Promise((resolve) => setTimeout(resolve, 100))
+			expect(enteredCriticalSection).toBe(false)
+
+			child.send("release")
+			await waitForMessage(child, "released")
+			await expect(blocked).resolves.toBe("acquired")
+			expect(enteredCriticalSection).toBe(true)
+		} finally {
+			if (!child.killed) {
+				child.kill()
+			}
+		}
+	})
+
+	it("reset is a no-op for file-based locking", () => {
+		const lock = new TaskHistoryLock()
+		expect(() => lock.reset()).not.toThrow()
+	})
+})