From 28834a91aeb31c4058bfa92558f9b45b5e5121c0 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Mon, 26 Oct 2026 23:13:25 +0700 Subject: [PATCH] fix(server): tolerate brief Windows atomic replace failures (#7955) A receipt could remain pending after provider delivery when Windows rejected the completed-state rename with EPERM. Retry transient replace errors within a one-second budget while retaining the original file and cleanup semantics. --- .../server/src/server/atomic-file.test.ts | 147 ++++++++++++++++++ packages/server/src/server/atomic-file.ts | 57 ++++++- 2 files changed, 201 insertions(+), 3 deletions(-) create mode 100644 packages/server/src/server/atomic-file.test.ts diff --git a/packages/server/src/server/atomic-file.test.ts b/packages/server/src/server/atomic-file.test.ts new file mode 100644 index 0000000000..a521a1cc3c --- /dev/null +++ b/packages/server/src/server/atomic-file.test.ts @@ -0,0 +1,147 @@ +import { promises as fs } from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, beforeEach, expect, it } from "vitest"; +import { writeFileAtomic, writeJsonFileAtomic } from "./atomic-file.js"; + +let directory: string; +beforeEach(async () => { + directory = await fs.mkdtemp(path.join(os.tmpdir(), "paseo-atomic-file-")); +}); +afterEach(async () => { + await fs.rm(directory, { recursive: true, force: true }); +}); + +it.each(["EPERM", "EACCES", "EBUSY"])( + "commits a completed receipt after transient Windows %s replace failures", + async (code) => { + const file = path.join(directory, "receipt.json"); + await writeJsonFileAtomic(file, { state: "pending" }); + let failures = 3; + let elapsed = 0; + await writeJsonFileAtomic( + file, + { state: "completed" }, + { + platform: "win32", + now: () => elapsed, + sleep: async (milliseconds) => { + elapsed += milliseconds; + }, + rename: async (source, destination) => { + if (failures-- > 0) { + expect(JSON.parse(await fs.readFile(destination, "utf8"))).toEqual({ + state: "pending", + }); + expect(JSON.parse(await fs.readFile(source, "utf8"))).toEqual({ state: "completed" }); + throw Object.assign(new Error("destination held open"), { code }); + } + await fs.rename(source, destination); + }, + }, + ); + expect(JSON.parse(await fs.readFile(file, "utf8"))).toEqual({ state: "completed" }); + expect(await fs.readdir(directory)).toEqual(["receipt.json"]); + expect(elapsed).toBeGreaterThan(0); + }, +); + +it.each(["EPERM", "EACCES", "EBUSY"])( + "preserves the previous file and removes the temp file when Windows %s persists", + async (code) => { + const file = path.join(directory, "receipt.json"); + await writeJsonFileAtomic(file, { state: "pending" }); + const error = Object.assign(new Error("destination held open"), { code }); + const waits: number[] = []; + let elapsed = 0; + const attempts: number[] = []; + await expect( + writeJsonFileAtomic( + file, + { state: "completed" }, + { + platform: "win32", + now: () => elapsed, + sleep: async (milliseconds) => { + waits.push(milliseconds); + elapsed += milliseconds; + }, + rename: async () => { + attempts.push(elapsed); + throw error; + }, + }, + ), + ).rejects.toBe(error); + expect(elapsed).toBe(1000); + expect(waits.slice(0, 5)).toEqual([10, 20, 40, 80, 100]); + expect(Math.max(...waits)).toBe(100); + expect(attempts.every((time) => time < 1000)).toBe(true); + expect(JSON.parse(await fs.readFile(file, "utf8"))).toEqual({ state: "pending" }); + expect(await fs.readdir(directory)).toEqual(["receipt.json"]); + }, +); + +it.each([ + ["linux", "EPERM"], + ["darwin", "EACCES"], + ["linux", "EBUSY"], + ["win32", "ENOENT"], + ["win32", "ENOSPC"], + ["win32", "EXDEV"], +] as const)("does not retry %s %s failures", async (platform, code) => { + const file = path.join(directory, "receipt.json"); + await writeJsonFileAtomic(file, { state: "pending" }); + const error = Object.assign(new Error("rename failed"), { code }); + let attempts = 0; + const waits: number[] = []; + await expect( + writeFileAtomic(file, "replacement", { + platform, + now: () => 0, + sleep: async (milliseconds) => { + waits.push(milliseconds); + }, + rename: async () => { + attempts++; + throw error; + }, + }), + ).rejects.toBe(error); + expect(attempts).toBe(1); + expect(waits).toEqual([]); + expect(JSON.parse(await fs.readFile(file, "utf8"))).toEqual({ state: "pending" }); + expect(await fs.readdir(directory)).toEqual(["receipt.json"]); +}); + +it("does not retry after a rename consumes the Windows retry budget", async () => { + const file = path.join(directory, "receipt.json"); + const error = Object.assign(new Error("slow rename failed"), { code: "EPERM" }); + let elapsed = 0; + const waits: number[] = []; + await expect( + writeFileAtomic(file, "replacement", { + platform: "win32", + now: () => elapsed, + sleep: async (milliseconds) => { + waits.push(milliseconds); + }, + rename: async () => { + elapsed += 1000; + throw error; + }, + }), + ).rejects.toBe(error); + expect(waits).toEqual([]); + expect(await fs.readdir(directory)).toEqual([]); +}); + +it("creates parent directories and replaces binary and JSON files through the default runtime", async () => { + const file = path.join(directory, "nested", "file"); + const bytes = Buffer.from([0, 255, 128, 1]); + await writeFileAtomic(file, bytes); + expect(await fs.readFile(file)).toEqual(bytes); + await writeJsonFileAtomic(file, { state: "completed" }); + expect(await fs.readFile(file, "utf8")).toBe('{\n "state": "completed"\n}'); + expect(await fs.readdir(path.dirname(file))).toEqual(["file"]); +}); diff --git a/packages/server/src/server/atomic-file.ts b/packages/server/src/server/atomic-file.ts index cedc8d51ea..9f9c9f9568 100644 --- a/packages/server/src/server/atomic-file.ts +++ b/packages/server/src/server/atomic-file.ts @@ -1,10 +1,26 @@ import { randomUUID } from "node:crypto"; import { promises as fs } from "node:fs"; import path from "node:path"; +import { setTimeout as sleep } from "node:timers/promises"; + +interface AtomicFileRuntime { + platform: NodeJS.Platform; + rename: (source: string, destination: string) => Promise; + now: () => number; + sleep: (milliseconds: number) => Promise; +} + +const runtime: AtomicFileRuntime = { + platform: process.platform, + rename: fs.rename, + now: () => performance.now(), + sleep, +}; export async function writeFileAtomic( filePath: string, data: string | NodeJS.ArrayBufferView, + operations: AtomicFileRuntime = runtime, ): Promise { await fs.mkdir(path.dirname(filePath), { recursive: true }); const tempPath = path.join( @@ -13,13 +29,48 @@ export async function writeFileAtomic( ); try { await fs.writeFile(tempPath, data, "utf8"); - await fs.rename(tempPath, filePath); + await replaceFile(tempPath, filePath, operations); } catch (error) { await fs.rm(tempPath, { force: true }); throw error; } } -export async function writeJsonFileAtomic(filePath: string, value: unknown): Promise { - await writeFileAtomic(filePath, JSON.stringify(value, null, 2)); +async function replaceFile( + source: string, + destination: string, + operations: AtomicFileRuntime, +): Promise { + // Windows readers/scanners can briefly deny replacement. Yield with backoff, + // but keep interactive persistence failures bounded to a one-second retry budget. + const deadline = operations.now() + 1000; + let delay = 10; + for (;;) { + try { + await operations.rename(source, destination); + return; + } catch (error) { + if ( + operations.platform !== "win32" || + !(error instanceof Error) || + !("code" in error) || + !["EPERM", "EACCES", "EBUSY"].includes(String(error.code)) + ) { + throw error; + } + const remaining = deadline - operations.now(); + if (remaining <= 0) throw error; + await operations.sleep(Math.min(delay, remaining)); + if (operations.now() >= deadline) throw error; + delay = Math.min(delay * 2, 100); + } + } +} + +export async function writeJsonFileAtomic( + filePath: string, + value: unknown, + operations: AtomicFileRuntime = runtime, +): Promise { + await writeFileAtomic(filePath, JSON.stringify(value, null, 2), operations); }