Skip to content

Commit 4e6fa84

Browse files
committed
fix(task-history): atomic read-modify-write
prevents cross-process lost updates
1 parent 2fe1189 commit 4e6fa84

5 files changed

Lines changed: 244 additions & 43 deletions

File tree

‎src/core/task-persistence/TaskHistoryStore.ts‎

Lines changed: 73 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -78,12 +78,11 @@ interface DelegationRepairIntent {
7878
* A single index file (`globalStorage/tasks/_index.json`) is maintained
7979
* as a cache for fast list reads at startup.
8080
*
81-
* Cross-process safety for per-task files comes from `safeWriteJson`'s
82-
* `proper-lockfile`. The shared `_index.json` is a best-effort startup
83-
* cache: it may lag or miss entries from other hosts, but
84-
* `reconcile()` rebuilds from authoritative per-task files — at startup
85-
* (`forceRefresh: true`), via `fs.watch`, and on a 5-minute periodic
86-
* fallback. Within a single extension host process,
81+
* Cross-process safety for per-task files and `_index.json` comes from
82+
* `safeWriteJson`'s `proper-lockfile` with a `merge` callback: each
83+
* write reads the current file under the advisory lock and merges
84+
* incoming fields, so a concurrent writer's changes are preserved
85+
* rather than silently dropped. Within a single extension host process,
8786
* an in-process write lock serializes mutations.
8887
*/
8988
/**
@@ -261,8 +260,15 @@ export class TaskHistoryStore {
261260
// Merge: preserve existing metadata unless explicitly overwritten
262261
const merged = existing ? { ...existing, ...item } : item
263262

264-
// Write per-task file (source of truth)
265-
await this.writeTaskFile(merged)
263+
// Compute the actual changed fields relative to the cached state.
264+
// Only these are applied to the disk version, so fields updated by
265+
// another process are preserved rather than reverted from a stale cache.
266+
const delta = existing
267+
? Object.fromEntries(
268+
Object.entries(item).filter(([k, v]) => !deepEqual(v, (existing as Record<string, unknown>)[k])),
269+
)
270+
: undefined
271+
await this.writeTaskFile(merged, delta ? ({ id: item.id, ...delta } as HistoryItem) : undefined)
266272

267273
// Update in-memory cache
268274
this.cache.set(merged.id, merged)
@@ -884,17 +890,46 @@ export class TaskHistoryStore {
884890
}
885891

886892
/**
887-
* Write the full index to disk.
893+
* Write the index to disk, merging entries from other hosts.
894+
*
895+
* Peer entries (task ids present on disk but absent from this host's
896+
* cache) are kept only if their task directory still exists, so a
897+
* local delete propagates instead of being undone by the merge.
888898
*/
889899
private async writeIndex(): Promise<void> {
890900
const indexPath = await this.getIndexPath()
901+
const tasksDir = await this.getTasksDir()
891902
const index: HistoryIndex = {
892903
version: 1,
893904
updatedAt: Date.now(),
894905
entries: this.getAll(),
895906
}
896907

897-
await safeWriteJson(indexPath, index)
908+
let onDiskIds: Set<string>
909+
try {
910+
const dirEntries = await fs.readdir(tasksDir)
911+
onDiskIds = new Set(dirEntries.filter((n) => !n.startsWith("_") && !n.startsWith(".")))
912+
} catch {
913+
onDiskIds = new Set()
914+
}
915+
916+
await safeWriteJson(indexPath, index, {
917+
merge: (existing, incoming) => {
918+
if (!existing || existing.version !== 1 || !Array.isArray(existing.entries)) {
919+
return incoming
920+
}
921+
const ourIds = new Set(incoming.entries.map((e: HistoryItem) => e.id))
922+
const peerEntries = existing.entries.filter(
923+
(e: HistoryItem) => e.id && !ourIds.has(e.id) && onDiskIds.has(e.id),
924+
)
925+
return {
926+
...incoming,
927+
entries: [...incoming.entries, ...peerEntries].sort(
928+
(a: HistoryItem, b: HistoryItem) => b.ts - a.ts,
929+
),
930+
}
931+
},
932+
})
898933
}
899934

900935
/**
@@ -935,10 +970,26 @@ export class TaskHistoryStore {
935970

936971
/**
937972
* Write a HistoryItem to its per-task `history_item.json` file.
973+
*
974+
* When `delta` is provided, the merge callback applies only the
975+
* delta to the current disk state, so fields written by another
976+
* process are preserved. Without a delta the full item is written
977+
* as-is (used by administrative repair paths that are authoritative).
938978
*/
939-
private async writeTaskFile(item: HistoryItem): Promise<void> {
979+
private async writeTaskFile(item: HistoryItem, delta?: Partial<HistoryItem>): Promise<void> {
940980
const filePath = await this.getTaskFilePath(item.id)
941-
await safeWriteJson(filePath, item)
981+
if (delta) {
982+
await safeWriteJson(filePath, item, {
983+
merge: (existing, incoming) => {
984+
if (!existing || typeof existing !== "object" || !("id" in existing)) {
985+
return incoming
986+
}
987+
return { ...existing, ...delta }
988+
},
989+
})
990+
} else {
991+
await safeWriteJson(filePath, item)
992+
}
942993
}
943994

944995
/**
@@ -1109,10 +1160,18 @@ export class TaskHistoryStore {
11091160
const mergedFirst = { ...first, ...updatedFirst }
11101161
const mergedSecond = { ...second, ...updatedSecond }
11111162

1163+
// Compute actual diffs against cached state, mirroring upsertCore.
1164+
const deltaFirst = Object.fromEntries(
1165+
Object.entries(updatedFirst).filter(([k, v]) => !deepEqual(v, (first as Record<string, unknown>)[k])),
1166+
)
1167+
const deltaSecond = Object.fromEntries(
1168+
Object.entries(updatedSecond).filter(([k, v]) => !deepEqual(v, (second as Record<string, unknown>)[k])),
1169+
)
1170+
11121171
// Write both files before touching the cache so readers never observe a
11131172
// half-updated in-memory state between the two await points.
1114-
await this.writeTaskFile(mergedFirst)
1115-
await this.writeTaskFile(mergedSecond)
1173+
await this.writeTaskFile(mergedFirst, { id: firstId, ...deltaFirst } as HistoryItem)
1174+
await this.writeTaskFile(mergedSecond, { id: secondId, ...deltaSecond } as HistoryItem)
11161175

11171176
// Both disk writes succeeded — now update the cache atomically.
11181177
this.cache.set(firstId, mergedFirst)

‎src/core/task-persistence/__tests__/TaskHistoryStore.crossInstance.spec.ts‎

Lines changed: 112 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -15,12 +15,30 @@ vi.mock("../../../utils/storage", () => ({
1515
}),
1616
}))
1717

18-
// Mock safeWriteJson to use plain fs writes in tests (avoids proper-lockfile issues)
18+
// Mock safeWriteJson to use plain fs writes but honor the merge callback.
1919
vi.mock("../../../utils/safeWriteJson", () => ({
20-
safeWriteJson: vi.fn().mockImplementation(async (filePath: string, data: any) => {
21-
await fs.mkdir(path.dirname(filePath), { recursive: true })
22-
await fs.writeFile(filePath, JSON.stringify(data, null, "\t"), "utf8")
23-
}),
20+
safeWriteJson: vi
21+
.fn()
22+
.mockImplementation(
23+
async (
24+
filePath: string,
25+
data: unknown,
26+
options?: { merge?: (existing: unknown, incoming: unknown) => unknown },
27+
) => {
28+
await fs.mkdir(path.dirname(filePath), { recursive: true })
29+
if (options?.merge) {
30+
let existing: unknown = null
31+
try {
32+
const raw = await fs.readFile(filePath, "utf8")
33+
existing = JSON.parse(raw)
34+
} catch {
35+
// File does not exist or is corrupt
36+
}
37+
data = options.merge(existing, data)
38+
}
39+
await fs.writeFile(filePath, JSON.stringify(data, null, "\t"), "utf8")
40+
},
41+
),
2442
}))
2543

2644
function makeHistoryItem(overrides: Partial<HistoryItem> = {}): HistoryItem {
@@ -166,12 +184,12 @@ describe("TaskHistoryStore cross-instance safety", () => {
166184
})
167185

168186
/**
169-
* Regression for #1231: two hosts each hold only their own task in cache and
170-
* flush `_index.json` without watcher reconciliation. The last writer's
171-
* index reflects only its own cache (LWW). Per-task files remain intact,
172-
* and reconcile recovers the full set.
187+
* Two hosts each hold only their own task in cache
188+
* and flush `_index.json` without watcher reconciliation. The merge
189+
* callback in writeIndex preserves peer entries so neither host's
190+
* flush drops the other's task.
173191
*/
174-
it("reconcile recovers peer entries after a last-writer-wins index flush", async () => {
192+
it("index flush merges peer entries instead of clobbering them", async () => {
175193
await storeA.initialize()
176194
await storeB.initialize()
177195

@@ -181,6 +199,7 @@ describe("TaskHistoryStore cross-instance safety", () => {
181199
await storeA.upsert(makeHistoryItem({ id: "task-a", task: "from A", ts: 1000 }))
182200
await storeB.upsert(makeHistoryItem({ id: "task-b", task: "from B", ts: 2000 }))
183201

202+
// Each cache is intentionally partial.
184203
expect(storeA.get("task-a")).toBeDefined()
185204
expect(storeA.get("task-b")).toBeUndefined()
186205
expect(storeB.get("task-b")).toBeDefined()
@@ -189,28 +208,97 @@ describe("TaskHistoryStore cross-instance safety", () => {
189208
await storeA.flushIndex()
190209
await storeB.flushIndex()
191210

192-
// Per-task files survive regardless of which host flushed last.
211+
// Both entries survive in the index — no reconcile needed.
193212
const tasksDir = path.join(tmpDir, "tasks")
213+
const indexRaw = await fs.readFile(path.join(tasksDir, GlobalFileNames.historyIndex), "utf8")
214+
const index = JSON.parse(indexRaw) as { entries: HistoryItem[] }
215+
const indexIds = index.entries.map((entry) => entry.id).sort()
216+
expect(indexIds).toEqual(["task-a", "task-b"])
217+
})
194218

195-
// The index reflects the last writer's partial cache (LWW clobber).
196-
const indexRawBefore = await fs.readFile(path.join(tasksDir, GlobalFileNames.historyIndex), "utf8")
197-
const indexBefore = JSON.parse(indexRawBefore) as { entries: HistoryItem[] }
198-
expect(indexBefore.entries.map((e) => e.id)).toEqual(["task-b"])
199-
const taskDirs = (await fs.readdir(tasksDir)).filter((name) => !name.startsWith("_") && !name.startsWith("."))
200-
expect(taskDirs.sort()).toEqual(["task-a", "task-b"])
219+
/**
220+
* A deleted task must not reappear in the index after a subsequent flush.
221+
* The merge keeps peer entries only if their task directory exists.
222+
*/
223+
it("index flush does not resurrect deleted tasks via the merge", async () => {
224+
await storeA.initialize()
225+
await storeB.initialize()
201226

202-
// Reconcile rebuilds the full picture from authoritative per-task files.
203-
await storeA.reconcile({ forceRefresh: true })
204-
expect(storeA.get("task-a")).toBeDefined()
205-
expect(storeA.get("task-b")).toBeDefined()
206-
expect(storeA.getAll()).toHaveLength(2)
227+
disableBackgroundReconciliation(storeA)
228+
disableBackgroundReconciliation(storeB)
229+
230+
await storeA.upsert(makeHistoryItem({ id: "keep", ts: 1000 }))
231+
await storeA.upsert(makeHistoryItem({ id: "doomed", ts: 2000 }))
232+
await storeA.flushIndex()
233+
234+
// Delete doomed — removes from cache and unlinks the file.
235+
await storeA.delete("doomed")
236+
const taskDir = path.join(tmpDir, "tasks", "doomed")
237+
await fs.rm(taskDir, { recursive: true, force: true })
207238

208-
// A post-reconcile flush writes the complete set.
239+
// Flush again — the merge must not re-add "doomed" from the old index.
209240
await storeA.flushIndex()
241+
242+
const tasksDir = path.join(tmpDir, "tasks")
210243
const indexRaw = await fs.readFile(path.join(tasksDir, GlobalFileNames.historyIndex), "utf8")
211244
const index = JSON.parse(indexRaw) as { entries: HistoryItem[] }
212245
const indexIds = index.entries.map((entry) => entry.id).sort()
213-
expect(indexIds).toEqual(["task-a", "task-b"])
246+
expect(indexIds).toEqual(["keep"])
247+
})
248+
249+
/**
250+
* Host B completes a task on disk while host A's cache still has it
251+
* active. Host A's next save updates only totalCost (a full-object
252+
* upsert — the realistic production shape). The diff-delta merge
253+
* preserves B's status because status did not change in A's cache.
254+
*/
255+
it("per-task diff-delta preserves a peer's status change on full-object upsert", async () => {
256+
await storeA.initialize()
257+
258+
// Base item with an explicit status — mirrors real production items.
259+
const base = makeHistoryItem({ id: "shared-task", status: "active", totalCost: 0.01, ts: 1000 })
260+
await storeA.upsert(base)
261+
262+
// Host B completes the task on disk; A's cache still has "active".
263+
const filePath = path.join(tmpDir, "tasks", "shared-task", GlobalFileNames.historyItem)
264+
const onDisk = JSON.parse(await fs.readFile(filePath, "utf8"))
265+
onDisk.status = "completed"
266+
onDisk.completionResultSummary = "done by host B"
267+
await fs.writeFile(filePath, JSON.stringify(onDisk), "utf8")
268+
269+
// Host A does a full-object upsert (the realistic path — spread the
270+
// cached item and change one field). The cached item has status: "active".
271+
await storeA.upsert({ ...storeA.get("shared-task")!, totalCost: 9.99 })
272+
273+
const final = JSON.parse(await fs.readFile(filePath, "utf8")) as HistoryItem
274+
expect(final.totalCost).toBe(9.99)
275+
// Status is preserved from disk because A's delta does not include
276+
// status — it was unchanged relative to A's cache.
277+
expect(final.status).toBe("completed")
278+
expect(final.completionResultSummary).toBe("done by host B")
279+
})
280+
281+
/**
282+
* When both hosts change the same field, the last writer wins.
283+
* This is expected — true conflict resolution requires application
284+
* semantics that a generic merge cannot provide.
285+
*/
286+
it("same-field changes from both hosts are last-writer-wins", async () => {
287+
await storeA.initialize()
288+
await storeB.initialize()
289+
290+
const base = makeHistoryItem({ id: "shared-task", status: "active", totalCost: 0.01, ts: 1000 })
291+
await storeA.upsert(base)
292+
await storeB.reconcile()
293+
294+
// Both hosts change totalCost.
295+
await storeA.upsert({ ...storeA.get("shared-task")!, totalCost: 1.0 })
296+
await storeB.upsert({ ...storeB.get("shared-task")!, totalCost: 2.0 })
297+
298+
const filePath = path.join(tmpDir, "tasks", "shared-task", GlobalFileNames.historyItem)
299+
const final = JSON.parse(await fs.readFile(filePath, "utf8")) as HistoryItem
300+
// B wrote last, so B's value wins.
301+
expect(final.totalCost).toBe(2.0)
214302
})
215303
})
216304

‎src/eslint-suppressions.json‎

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -779,11 +779,6 @@
779779
"count": 4
780780
}
781781
},
782-
"core/task-persistence/__tests__/TaskHistoryStore.crossInstance.spec.ts": {
783-
"@typescript-eslint/no-explicit-any": {
784-
"count": 1
785-
}
786-
},
787782
"core/task-persistence/__tests__/TaskHistoryStore.spec.ts": {
788783
"@typescript-eslint/no-explicit-any": {
789784
"count": 4

‎src/utils/__tests__/safeWriteJson.test.ts‎

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -468,4 +468,40 @@ describe("safeWriteJson", () => {
468468

469469
consoleErrorSpy.mockRestore()
470470
})
471+
472+
// Merge option tests
473+
test("should merge incoming data with existing file content when merge callback is provided", async () => {
474+
const initial = { a: 1, b: 2 }
475+
await safeWriteJson(currentTestFilePath, initial)
476+
477+
const incoming = { b: 3, c: 4 }
478+
await safeWriteJson(currentTestFilePath, incoming, {
479+
merge: (existing, data) => ({ ...existing, ...data }),
480+
})
481+
482+
const content = await readFileContent(currentTestFilePath)
483+
expect(content).toEqual({ a: 1, b: 3, c: 4 })
484+
})
485+
486+
test("should pass null to merge callback when file does not exist", async () => {
487+
const newFilePath = path.join(tempDir, "nonexistent.json")
488+
const mergeFn = vi.fn((existing, incoming) => incoming)
489+
490+
await safeWriteJson(newFilePath, { value: 42 }, { merge: mergeFn })
491+
492+
expect(mergeFn).toHaveBeenCalledWith(null, { value: 42 })
493+
const content = await readFileContent(newFilePath)
494+
expect(content).toEqual({ value: 42 })
495+
})
496+
497+
test("should write incoming data directly when no merge callback is provided", async () => {
498+
const initial = { a: 1, b: 2 }
499+
await safeWriteJson(currentTestFilePath, initial)
500+
501+
const replacement = { c: 3 }
502+
await safeWriteJson(currentTestFilePath, replacement)
503+
504+
const content = await readFileContent(currentTestFilePath)
505+
expect(content).toEqual({ c: 3 })
506+
})
471507
})

‎src/utils/safeWriteJson.ts‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,16 @@ export interface SafeWriteJsonOptions {
1515
* @default false
1616
*/
1717
prettyPrint?: boolean
18+
19+
/**
20+
* When provided, the current file is read under the advisory lock
21+
* and passed to this function along with the incoming data. The
22+
* return value replaces `data` for the write. This turns a blind
23+
* overwrite into an atomic read-modify-write, preventing cross-process
24+
* lost updates. `existing` is null when the file does not exist or
25+
* cannot be parsed.
26+
*/
27+
merge?: (existing: unknown, incoming: unknown) => unknown
1828
}
1929

2030
/**
@@ -78,6 +88,19 @@ async function safeWriteJson(filePath: string, data: any, options?: SafeWriteJso
7888
throw lockError
7989
}
8090

91+
// If a merge callback was provided, read the current file under the lock
92+
// and let the caller merge before we write.
93+
if (options?.merge) {
94+
let existing: unknown = null
95+
try {
96+
const raw = await fs.readFile(absoluteFilePath, "utf8")
97+
existing = JSON.parse(raw)
98+
} catch {
99+
// No readable file yet, so the merge receives null.
100+
}
101+
data = options.merge(existing, data)
102+
}
103+
81104
// Variables to hold the actual paths of temp files if they are created.
82105
let actualTempNewFilePath: string | null = null
83106
let actualTempBackupFilePath: string | null = null

0 commit comments

Comments
 (0)