Skip to content

Commit b66edb1

Browse files
easonliang28easonliang28
committed
fix: propagate non-ENOENT tail-inspection errors and complete partial writes
appendChange previously swallowed every tail-inspection error, so an I/O failure could let the append proceed without the required separator and silently lose the entry to a merged line; only ENOENT (fresh journal) is suppressed now. The append also ignored FileHandle.write's bytesWritten, so a partial write became a durable torn record reported as success; the loop now writes the whole record before syncing. Regression tests cover both paths.
1 parent eae754b commit b66edb1

2 files changed

Lines changed: 69 additions & 4 deletions

File tree

‎src/core/checkpoints/__tests__/changeJournal.spec.ts‎

Lines changed: 54 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@ import fs from "fs/promises"
22
import os from "os"
33
import path from "path"
44

5-
import { afterEach, beforeEach, describe, expect, it } from "vitest"
5+
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"
66

77
import { appendChange, journalPath, loadChanges, type ChangeJournalEntry } from "../changeJournal"
88

@@ -71,6 +71,59 @@ describe("changeJournal", () => {
7171
expect(raw).not.toContain("\n\n")
7272
expect(raw.split("\n").filter((line) => line !== "")).toHaveLength(2)
7373
})
74+
75+
it("propagates a non-ENOENT tail-inspection failure instead of appending blind", async () => {
76+
await appendChange(tmpRoot, taskId, entry({ checkpointId: "good" }))
77+
const filePath = journalPath(tmpRoot, taskId)
78+
const raw = await fs.readFile(filePath, "utf8")
79+
await fs.writeFile(filePath, raw + '{"path":"src/half.ts","operation":"upd')
80+
// The tail cannot be inspected (I/O error): appending would merge the
81+
// new entry into the fragment, so appendChange must fail loudly.
82+
const statSpy = vi
83+
.spyOn(fs, "stat")
84+
.mockRejectedValueOnce(Object.assign(new Error("simulated I/O error"), { code: "EIO" }))
85+
try {
86+
await expect(appendChange(tmpRoot, taskId, entry({ checkpointId: "after" }))).rejects.toMatchObject({
87+
code: "EIO",
88+
})
89+
} finally {
90+
statSpy.mockRestore()
91+
}
92+
// The failed append wrote nothing: the next append still repairs and loads.
93+
await appendChange(tmpRoot, taskId, entry({ checkpointId: "after" }))
94+
const result = await loadChanges(tmpRoot, taskId)
95+
expect(result.map((e) => e.checkpointId)).toEqual(["good", "after"])
96+
})
97+
98+
it("writes the full record when the underlying write completes partially", async () => {
99+
await appendChange(tmpRoot, taskId, entry({ checkpointId: "first" }))
100+
const realOpen = fs.open
101+
const openSpy = vi.spyOn(fs, "open").mockImplementation(async (pathish, flags) => {
102+
const openAs = realOpen as (p: string | URL, f?: string) => Promise<fs.FileHandle>
103+
const handle = await openAs(pathish as string | URL, flags as string | undefined)
104+
if (flags === "a") {
105+
const realWrite = handle.write.bind(handle)
106+
let first = true
107+
handle.write = (async (buf: Uint8Array, off?: number, len?: number) => {
108+
const length = len ?? buf.length
109+
if (first && length > 1) {
110+
first = false
111+
await realWrite(buf, off, 1)
112+
return { bytesWritten: 1, buffer: Buffer.from(buf) }
113+
}
114+
return realWrite(buf, off, length)
115+
}) as typeof handle.write
116+
}
117+
return handle
118+
})
119+
try {
120+
await appendChange(tmpRoot, taskId, entry({ checkpointId: "second" }))
121+
} finally {
122+
openSpy.mockRestore()
123+
}
124+
const result = await loadChanges(tmpRoot, taskId)
125+
expect(result.map((e) => e.checkpointId)).toEqual(["first", "second"])
126+
})
74127
})
75128

76129
describe("task ID containment", () => {

‎src/core/checkpoints/changeJournal.ts‎

Lines changed: 15 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -76,13 +76,25 @@ export async function appendChange(globalStorageDir: string, taskId: string, ent
7676
await tail.close()
7777
}
7878
}
79-
} catch {
80-
// Journal absent (fresh append) or vanished between stat and open.
79+
} catch (error) {
80+
// A missing journal is a fresh append. Any other inspection failure
81+
// means we cannot know whether a separator is required, so propagate
82+
// instead of appending blind and losing the entry to a merged line.
83+
if ((error as NodeJS.ErrnoException).code !== "ENOENT") {
84+
throw error
85+
}
8186
}
8287
const data = Buffer.from(payload)
8388
const handle = await fs.open(filePath, "a")
8489
try {
85-
await handle.write(data, 0, data.length)
90+
// A single write can complete partially; sync only after the whole
91+
// record is on the handle, or a short write would persist a torn entry
92+
// while appendChange reports success.
93+
let written = 0
94+
while (written < data.length) {
95+
const res = await handle.write(data, written, data.length - written)
96+
written += res.bytesWritten
97+
}
8698
await handle.sync()
8799
} finally {
88100
await handle.close()

0 commit comments

Comments
 (0)