root/fusion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(HAI-020): add atomic writes and remove truncation recovery workarounds

Browse Source 
- Add atomic writes for config.json with config lock in store
- Simplify safeReadTaskJson to readTaskJson, removing truncation recovery logic
- Remove withRetry workaround from fetchTaskDetail in dashboard API
- Remove related retry logic from executor and dashboard command
- Update store tests to cover atomic write behavior
This commit is contained in:
Dustin Byrne
2026-03-25 21:43:57 -04:00
parent 99502e4e05 2488e1c9d2
commit f34d849282
3 changed files with 97 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
packages/core/src/store.test.ts
View File

@@ -90,19 +90,17 @@ describe("TaskStore", () => {
// ── Defensive parsing test ───────────────────────────────────────
describe("defensive JSON parsing", () => {
it("recovers from corrupted task.json with trailing duplicate content", async () => {
it("throws on corrupted task.json with trailing duplicate content (atomic writes prevent this)", async () => {
const task = await createTestTask();
const taskJsonPath = join(rootDir, ".hai", "tasks", task.id, "task.json");
// Corrupt the file: append duplicate trailing content (like HAI-015)
// Corrupt the file: append duplicate trailing content
const validJson = await readFile(taskJsonPath, "utf-8");
const corrupted = validJson + validJson.slice(validJson.length / 2);
await writeFile(taskJsonPath, corrupted);
// getTask should recover
const recovered = await store.getTask(task.id);
expect(recovered.id).toBe(task.id);
expect(recovered.description).toBe("Test task");
// With atomic writes, corruption indicates a real bug — should throw
await expect(store.getTask(task.id)).rejects.toThrow("Failed to parse task.json");
});
it("throws a clear error when JSON is completely unrecoverable", async () => {
@@ -137,6 +135,36 @@ describe("TaskStore", () => {
});
});
// ── Atomic config writes ──────────────────────────────────────────
describe("atomic config writes", () => {
it("produces valid config.json with unique sequential IDs after 5 parallel createTask calls", async () => {
const promises = Array.from({ length: 5 }, (_, i) =>
store.createTask({ description: `Concurrent task ${i}` }),
);
const tasks = await Promise.all(promises);
// All IDs should be unique
const ids = tasks.map((t) => t.id);
expect(new Set(ids).size).toBe(5);
// IDs should be sequential (HAI-001 through HAI-005)
const sortedIds = [...ids].sort();
expect(sortedIds).toEqual(["HAI-001", "HAI-002", "HAI-003", "HAI-004", "HAI-005"]);
// config.json should be valid JSON with nextId = 6
const configPath = join(rootDir, ".hai", "config.json");
const raw = await readFile(configPath, "utf-8");
const config = JSON.parse(raw);
expect(config.nextId).toBe(6);
// No .tmp files left behind
const haiDir = join(rootDir, ".hai");
const files = await readdir(haiDir);
expect(files.filter((f) => f.endsWith(".tmp"))).toHaveLength(0);
});
});
// ── Concurrent stress test ───────────────────────────────────────
describe("concurrent stress", () => {

103
packages/core/src/store.ts
View File

@@ -31,6 +31,8 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
private debounceMs = 150;
/** Per-task promise chain for serializing writes */
private taskLocks: Map<string, Promise<void>> = new Map();
/** Promise chain for serializing config.json read-modify-write cycles */
private configLock: Promise<void> = Promise.resolve();
constructor(private rootDir: string) {
super();
@@ -46,6 +48,26 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
}
}
/**
* Serialize all mutations to config.json by chaining promises.
* Concurrent callers will queue behind each other, preventing
* lost-update races on the nextId counter.
*/
private withConfigLock<T>(fn: () => Promise<T>): Promise<T> {
let resolve: () => void;
const next = new Promise<void>((r) => { resolve = r; });
const prev = this.configLock;
this.configLock = next;
return prev.then(async () => {
try {
return await fn();
} finally {
resolve!();
}
});
}
/**
* Serialize all mutations to a given task's task.json by chaining promises
* per task ID. Concurrent callers for the same ID will queue behind each other.
@@ -69,32 +91,16 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
}
/**
* Safely read and parse a task.json file. On `SyntaxError`, attempts to
* recover by truncating the content at the last valid `}` and re-parsing.
* Logs a warning to stderr when truncation-repair is used.
* Read and parse a task.json file. Throws immediately on invalid JSON —
* atomic writes (write-to-temp-then-rename) prevent partial-write
* corruption, so a `SyntaxError` indicates a real bug rather than a race.
*/
private async safeReadTaskJson(dir: string): Promise<Task> {
private async readTaskJson(dir: string): Promise<Task> {
const filePath = join(dir, "task.json");
const raw = await readFile(filePath, "utf-8");
try {
return JSON.parse(raw) as Task;
} catch (err) {
if (!(err instanceof SyntaxError)) throw err;
// Attempt recovery: try truncating at each '}' from the end until valid
let pos = raw.length;
while ((pos = raw.lastIndexOf("}", pos - 1)) > 0) {
try {
const task = JSON.parse(raw.slice(0, pos + 1)) as Task;
console.warn(
`[hai] Warning: repaired corrupted task.json at ${filePath} (truncated ${raw.length - pos - 1} trailing bytes)`,
);
return task;
} catch {
// Try next position
}
}
throw new Error(
`Failed to parse task.json at ${filePath}: ${(err as Error).message}`,
);
@@ -120,12 +126,14 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
}
async updateSettings(patch: Partial<Settings>): Promise<Settings> {
const config = await this.readConfig();
const current = { ...DEFAULT_SETTINGS, ...config.settings };
const updated = { ...current, ...patch };
config.settings = updated;
await this.writeConfig(config);
return updated;
return this.withConfigLock(async () => {
const config = await this.readConfig();
const current = { ...DEFAULT_SETTINGS, ...config.settings };
const updated = { ...current, ...patch };
config.settings = updated;
await this.writeConfig(config);
return updated;
});
}
private async readConfig(): Promise<BoardConfig> {
@@ -133,16 +141,29 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
return JSON.parse(data);
}
/**
* Atomically write config.json by writing to a temp file first, then
* renaming into place. The rename is atomic on POSIX filesystems,
* preventing partial writes from corrupting the file.
*/
private async atomicWriteConfig(config: BoardConfig): Promise<void> {
const tmpPath = this.configPath + ".tmp";
await writeFile(tmpPath, JSON.stringify(config, null, 2));
await rename(tmpPath, this.configPath);
}
private async writeConfig(config: BoardConfig): Promise<void> {
await writeFile(this.configPath, JSON.stringify(config, null, 2));
await this.atomicWriteConfig(config);
}
private async allocateId(): Promise<string> {
const config = await this.readConfig();
const id = `HAI-${String(config.nextId).padStart(3, "0")}`;
config.nextId++;
await this.writeConfig(config);
return id;
return this.withConfigLock(async () => {
const config = await this.readConfig();
const id = `HAI-${String(config.nextId).padStart(3, "0")}`;
config.nextId++;
await this.writeConfig(config);
return id;
});
}
private taskDir(id: string): string {
@@ -195,7 +216,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
*/
async getTask(id: string): Promise<TaskDetail> {
const dir = this.taskDir(id);
const task = await this.safeReadTaskJson(dir);
const task = await this.readTaskJson(dir);
let prompt = "";
const promptPath = join(dir, "PROMPT.md");
@@ -215,7 +236,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
for (const entry of entries) {
if (entry.isDirectory() && entry.name.startsWith("HAI-")) {
try {
tasks.push(await this.safeReadTaskJson(join(this.tasksDir, entry.name)));
tasks.push(await this.readTaskJson(join(this.tasksDir, entry.name)));
} catch {
// skip invalid task dirs
}
@@ -228,7 +249,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
async moveTask(id: string, toColumn: Column): Promise<Task> {
return this.withTaskLock(id, async () => {
const dir = this.taskDir(id);
const task = await this.safeReadTaskJson(dir);
const task = await this.readTaskJson(dir);
const validTargets = VALID_TRANSITIONS[task.column];
if (!validTargets.includes(toColumn)) {
@@ -264,7 +285,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
): Promise<Task> {
return this.withTaskLock(id, async () => {
const dir = this.taskDir(id);
const task = await this.safeReadTaskJson(dir);
const task = await this.readTaskJson(dir);
if (updates.title !== undefined) task.title = updates.title;
if (updates.description !== undefined) task.description = updates.description;
@@ -300,7 +321,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
): Promise<Task> {
return this.withTaskLock(id, async () => {
const dir = this.taskDir(id);
const task = await this.safeReadTaskJson(dir);
const task = await this.readTaskJson(dir);
// Auto-initialize steps from PROMPT.md if empty
if (task.steps.length === 0) {
@@ -348,7 +369,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
async logEntry(id: string, action: string, outcome?: string): Promise<Task> {
return this.withTaskLock(id, async () => {
const dir = this.taskDir(id);
const task = await this.safeReadTaskJson(dir);
const task = await this.readTaskJson(dir);
task.log.push({
timestamp: new Date().toISOString(),
@@ -415,7 +436,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
async deleteTask(id: string): Promise<Task> {
return this.withTaskLock(id, async () => {
const dir = this.taskDir(id);
const task = await this.safeReadTaskJson(dir);
const task = await this.readTaskJson(dir);
const taskJsonPath = join(dir, "task.json");
this.suppressWatcher(taskJsonPath);
@@ -438,7 +459,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
async mergeTask(id: string): Promise<MergeResult> {
return this.withTaskLock(id, async () => {
const dir = this.taskDir(id);
const task = await this.safeReadTaskJson(dir);
const task = await this.readTaskJson(dir);
if (task.column !== "in-review") {
throw new Error(
@@ -661,7 +682,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
let task: Task;
try {
const taskDir = join(this.tasksDir, taskId);
task = await this.safeReadTaskJson(taskDir);
task = await this.readTaskJson(taskDir);
} catch {
return; // File not readable or invalid JSON
}

17
packages/dashboard/app/api.ts
View File

@@ -10,27 +10,12 @@ async function api<T = unknown>(path: string, opts: RequestInit = {}): Promise<T
return data as T;
}
/**
* Retry wrapper for API calls that may fail due to transient server errors
* (e.g. 500s caused by concurrent file writes racing with reads).
* Retries once after a short delay before giving up.
*/
async function withRetry<T>(fn: () => Promise<T>, { retries = 1, delayMs = 200 } = {}): Promise<T> {
try {
return await fn();
} catch (err) {
if (retries <= 0) throw err;
await new Promise((r) => setTimeout(r, delayMs));
return withRetry(fn, { retries: retries - 1, delayMs });
}
}
export function fetchTasks(): Promise<Task[]> {
return api<Task[]>("/tasks");
}
export function fetchTaskDetail(id: string): Promise<TaskDetail> {
return withRetry(() => api<TaskDetail>(`/tasks/${id}`));
return api<TaskDetail>(`/tasks/${id}`);
}
export function createTask(input: TaskCreateInput): Promise<Task> {