stabilize runtime state on native Windows

This commit is contained in:
Sinabina
2026-07-17 14:21:10 -07:00
parent 9b3188e74b
commit 9bb382f2ee
8 changed files with 129 additions and 33 deletions
+2 -1
View File
@@ -531,7 +531,8 @@ export async function validateRuntimeBundle(directory, context = {}) {
const stat = await fs.lstat(absolute).catch(() => null); const stat = await fs.lstat(absolute).catch(() => null);
if (!stat?.isFile()) throw installError(`Runtime bundle file is missing: ${relative}`, "INSTALL_VALIDATION_FAILED"); if (!stat?.isFile()) throw installError(`Runtime bundle file is missing: ${relative}`, "INSTALL_VALIDATION_FAILED");
const digest = await sha256File(absolute); const digest = await sha256File(absolute);
if (digest !== file.sha256 || stat.size !== file.size || (stat.mode & 0o777) !== file.mode) { const modeMatches = (context.platform ?? process.platform) === "win32" || (stat.mode & 0o777) === file.mode;
if (digest !== file.sha256 || stat.size !== file.size || !modeMatches) {
throw installError(`Runtime bundle file failed integrity validation: ${relative}`, "INSTALL_VALIDATION_FAILED"); throw installError(`Runtime bundle file failed integrity validation: ${relative}`, "INSTALL_VALIDATION_FAILED");
} }
} }
+66 -17
View File
@@ -5,6 +5,7 @@ import path from "node:path";
import { randomUUID } from "node:crypto"; import { randomUUID } from "node:crypto";
const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms)); const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
const WINDOWS_TRANSIENT_FS_ERRORS = new Set(["EACCES", "EBUSY", "EPERM"]);
export async function pathExists(file) { export async function pathExists(file) {
try { try {
@@ -83,29 +84,65 @@ async function replaceFile(source, destination) {
} }
} }
async function syncDirectory(directory) { export async function syncDirectory(directory, options = {}) {
// Directory fsync is supported on Unix and not consistently on Windows. // Directory fsync is supported on Unix and not consistently on Windows.
const open = options.open ?? fs.open;
let handle;
let operationError;
try { try {
const handle = await fs.open(directory, "r"); handle = await open(directory, "r");
await handle.sync(); await handle.sync();
} catch (error) {
operationError = error;
}
let closeError;
if (handle) {
try {
await handle.close(); await handle.close();
} catch (error) { } catch (error) {
if (!(["EINVAL", "ENOTSUP", "EISDIR", "EPERM", "EACCES"].includes(error?.code))) { closeError = error;
throw error;
} }
} }
const unsupported = operationError && ["EINVAL", "ENOTSUP", "EISDIR", "EPERM", "EACCES"].includes(operationError?.code);
if (operationError && !unsupported) {
if (closeError) throw new AggregateError([operationError, closeError], `Directory sync and close failed: ${directory}`);
throw operationError;
}
if (closeError) throw closeError;
} }
export async function acquireLock(lockPath, options = {}) { export async function acquireLock(lockPath, options = {}) {
const timeoutMs = options.timeoutMs ?? 10_000; const timeoutMs = options.timeoutMs ?? 10_000;
const staleMs = options.staleMs ?? 120_000; const staleMs = options.staleMs ?? 120_000;
const platform = options.platform ?? process.platform;
const mkdir = options.mkdir ?? fs.mkdir;
const started = Date.now(); const started = Date.now();
const token = randomUUID(); const token = randomUUID();
await fs.mkdir(path.dirname(lockPath), { recursive: true, mode: 0o700 }); await mkdir(path.dirname(lockPath), { recursive: true, mode: 0o700 });
for (let attempt = 0; ; attempt += 1) { for (let attempt = 0; ; attempt += 1) {
let mkdirError;
try {
await mkdir(lockPath, { mode: 0o700 });
} catch (error) {
mkdirError = error;
}
if (mkdirError) {
const contended = mkdirError?.code === "EEXIST";
const windowsDeleteRace = platform === "win32" && mkdirError?.code === "EPERM";
if (!contended && !windowsDeleteRace) throw mkdirError;
if (contended) await reapStaleLock(lockPath, staleMs, platform);
if (Date.now() - started >= timeoutMs) {
const timeout = new Error(`Timed out waiting for lock ${lockPath}`);
timeout.code = "LOCK_TIMEOUT";
throw timeout;
}
const delay = Math.min(20, 2 + Math.floor(attempt / 3));
await sleep(delay);
continue;
}
try { try {
await fs.mkdir(lockPath, { mode: 0o700 });
const owner = { token, pid: process.pid, hostname: os.hostname(), createdAt: new Date().toISOString() }; const owner = { token, pid: process.pid, hostname: os.hostname(), createdAt: new Date().toISOString() };
await atomicWriteJson(path.join(lockPath, "owner.json"), owner, { mode: 0o600 }); await atomicWriteJson(path.join(lockPath, "owner.json"), owner, { mode: 0o600 });
const heartbeatMs = Math.max(1_000, Math.min(30_000, Math.floor(staleMs / 3))); const heartbeatMs = Math.max(1_000, Math.min(30_000, Math.floor(staleMs / 3)));
@@ -125,35 +162,47 @@ export async function acquireLock(lockPath, options = {}) {
if (current?.token === token) await fs.rm(lockPath, { recursive: true, force: true }); if (current?.token === token) await fs.rm(lockPath, { recursive: true, force: true });
}; };
} catch (error) { } catch (error) {
if (error?.code !== "EEXIST") throw error; await fs.rm(lockPath, { recursive: true, force: true }).catch(() => {});
await reapStaleLock(lockPath, staleMs); throw error;
if (Date.now() - started >= timeoutMs) {
const timeout = new Error(`Timed out waiting for lock ${lockPath}`);
timeout.code = "LOCK_TIMEOUT";
throw timeout;
}
const delay = Math.min(20, 2 + Math.floor(attempt / 3));
await sleep(delay);
} }
} }
} }
async function reapStaleLock(lockPath, staleMs) { async function reapStaleLock(lockPath, staleMs, platform = process.platform) {
try { try {
const stat = await fs.stat(lockPath); const stat = await fs.stat(lockPath);
if (Date.now() - stat.mtimeMs <= staleMs) return false; if (Date.now() - stat.mtimeMs <= staleMs) return false;
const owner = await readJson(path.join(lockPath, "owner.json"), null).catch(() => null); const owner = await readJson(path.join(lockPath, "owner.json"), null).catch(() => null);
if (owner?.hostname === os.hostname() && processIsAlive(owner.pid)) return false; if (owner?.hostname === os.hostname() && processIsAlive(owner.pid)) return false;
const staleName = `${lockPath}.stale-${process.pid}-${randomUUID()}`; const staleName = `${lockPath}.stale-${process.pid}-${randomUUID()}`;
await fs.rename(lockPath, staleName); await renameWithRetry(lockPath, staleName, { platform });
await fs.rm(staleName, { recursive: true, force: true }); await fs.rm(staleName, { recursive: true, force: true });
return true; return true;
} catch (error) { } catch (error) {
if (["ENOENT", "EEXIST", "ENOTEMPTY"].includes(error?.code)) return false; if (["ENOENT", "EEXIST", "ENOTEMPTY"].includes(error?.code)) return false;
if (platform === "win32" && error?.code === "EPERM") return false;
throw error; throw error;
} }
} }
/** Retry Windows rename races without weakening permanent errors elsewhere. */
export async function renameWithRetry(source, destination, options = {}) {
const platform = options.platform ?? process.platform;
const rename = options.rename ?? fs.rename;
const timeoutMs = options.timeoutMs ?? 2_000;
const started = Date.now();
for (let attempt = 0; ; attempt += 1) {
try {
return await rename(source, destination);
} catch (error) {
if (platform !== "win32" || !WINDOWS_TRANSIENT_FS_ERRORS.has(error?.code) || Date.now() - started >= timeoutMs) {
throw error;
}
await sleep(Math.min(50, 5 + attempt * 5));
}
}
}
function processIsAlive(pid) { function processIsAlive(pid) {
if (!Number.isInteger(pid) || pid <= 0) return false; if (!Number.isInteger(pid) || pid <= 0) return false;
try { try {
+4 -4
View File
@@ -2,7 +2,7 @@ import fs from "node:fs/promises";
import path from "node:path"; import path from "node:path";
import { randomUUID } from "node:crypto"; import { randomUUID } from "node:crypto";
import { assertPathInside, resolveRuntimePaths } from "./paths.js"; import { assertPathInside, resolveRuntimePaths } from "./paths.js";
import { atomicWriteJson, pathExists, readJson } from "./storage.js"; import { atomicWriteJson, pathExists, readJson, renameWithRetry } from "./storage.js";
import { import {
assertManagedHome, assertManagedHome,
ensureManagedHome, ensureManagedHome,
@@ -55,7 +55,7 @@ export async function stageUpgradeUnlocked(options) {
}, { mode: 0o644 }); }, { mode: 0o644 });
await assertTreeContainsNoLinks(stage); await assertTreeContainsNoLinks(stage);
if (options.verify) await options.verify(stage); if (options.verify) await options.verify(stage);
await fs.rename(stage, destination); await renameWithRetry(stage, destination);
staged = true; staged = true;
} catch (error) { } catch (error) {
await fs.rm(stage, { recursive: true, force: true }).catch(() => {}); await fs.rm(stage, { recursive: true, force: true }).catch(() => {});
@@ -263,7 +263,7 @@ export async function purgeManagedHomeUnlocked(home) {
for (const entry of present.filter((name) => managedEntries.has(name) && !preexisting.has(name))) { for (const entry of present.filter((name) => managedEntries.has(name) && !preexisting.has(name))) {
const source = assertPathInside(resolved, path.join(resolved, entry)); const source = assertPathInside(resolved, path.join(resolved, entry));
const destination = assertPathInside(quarantine, path.join(quarantine, entry)); const destination = assertPathInside(quarantine, path.join(quarantine, entry));
await fs.rename(source, destination); await renameWithRetry(source, destination);
moved.push({ source, destination }); moved.push({ source, destination });
} }
await fs.rm(quarantine, { recursive: true, force: true }); await fs.rm(quarantine, { recursive: true, force: true });
@@ -273,7 +273,7 @@ export async function purgeManagedHomeUnlocked(home) {
return { purged: true, home: resolved, preserved }; return { purged: true, home: resolved, preserved };
} catch (error) { } catch (error) {
for (const item of moved.reverse()) { for (const item of moved.reverse()) {
await fs.rename(item.destination, item.source).catch(() => {}); await renameWithRetry(item.destination, item.source).catch(() => {});
} }
await fs.rmdir(quarantine).catch(() => {}); await fs.rmdir(quarantine).catch(() => {});
throw error; throw error;
@@ -7,8 +7,10 @@ import {
cleanupRuntime, cleanupRuntime,
ensureManagedHome, ensureManagedHome,
pathExists, pathExists,
renameWithRetry,
resolveRuntimePaths, resolveRuntimePaths,
runtimeLifecycleLockPath, runtimeLifecycleLockPath,
syncDirectory,
} from "../runtime/index.js"; } from "../runtime/index.js";
const roots: string[] = []; const roots: string[] = [];
@@ -31,6 +33,44 @@ afterEach(async () => {
}); });
describe("runtime cleanup boundary", () => { describe("runtime cleanup boundary", () => {
test("retries transient Windows lock creation and rename races", async () => {
const home = await temporaryHome();
const lockPath = path.join(home, "locks", "windows-race.lock");
let mkdirAttempts = 0;
const release = await acquireLock(lockPath, {
platform: "win32",
mkdir: async (target: string, options: Record<string, unknown>) => {
if (target === lockPath && mkdirAttempts++ === 0) {
throw Object.assign(new Error("simulated Windows delete race"), { code: "EPERM" });
}
return fs.mkdir(target, options);
},
});
expect(mkdirAttempts).toBe(2);
await release();
let renameAttempts = 0;
await renameWithRetry("source", "destination", {
platform: "win32",
timeoutMs: 100,
rename: async () => {
if (renameAttempts++ < 2) throw Object.assign(new Error("simulated scanner race"), { code: "EPERM" });
},
});
expect(renameAttempts).toBe(3);
});
test("closes a directory handle when Windows does not support directory fsync", async () => {
let closes = 0;
await expect(syncDirectory("fixture", {
open: async () => ({
sync: async () => { throw Object.assign(new Error("unsupported directory sync"), { code: "EPERM" }); },
close: async () => { closes += 1; },
}),
})).resolves.toBeUndefined();
expect(closes).toBe(1);
});
test("removes only allowlisted stale runtime scratch and dead locks", async () => { test("removes only allowlisted stale runtime scratch and dead locks", async () => {
const home = await temporaryHome(); const home = await temporaryHome();
const paths = resolveRuntimePaths({ home }); const paths = resolveRuntimePaths({ home });
+6 -3
View File
@@ -31,8 +31,10 @@ async function temporaryRoot(label = "gstack2 runtime ") {
} }
afterEach(async () => { afterEach(async () => {
await Promise.all(temporaryRoots.splice(0).map((root) => for (const root of temporaryRoots.splice(0)) {
fs.chmod(root, 0o700).catch(() => {}).then(() => fs.rm(root, { recursive: true, force: true })))); await fs.chmod(root, 0o700).catch(() => {});
await fs.rm(root, { recursive: true, force: true, maxRetries: 5, retryDelay: 50 });
}
}); });
describe("gstack 2 host-neutral paths and state", () => { describe("gstack 2 host-neutral paths and state", () => {
@@ -92,10 +94,11 @@ describe("gstack 2 host-neutral paths and state", () => {
gitDir: path.join(root, "repo", ".git"), gitDir: path.join(root, "repo", ".git"),
}); });
await initializeProject(home, identity); await initializeProject(home, identity);
await Promise.all(Array.from({ length: 60 }, () => const updates = await Promise.allSettled(Array.from({ length: 60 }, () =>
updateProjectState(home, identity.projectId, (state) => { updateProjectState(home, identity.projectId, (state) => {
state.concurrentCounter = Number(state.concurrentCounter ?? 0) + 1; state.concurrentCounter = Number(state.concurrentCounter ?? 0) + 1;
}))); })));
expect(updates.filter((result) => result.status === "rejected")).toEqual([]);
const { state } = await inspectProject(home, identity); const { state } = await inspectProject(home, identity);
expect(state.concurrentCounter).toBe(60); expect(state.concurrentCounter).toBe(60);
expect(state.revision).toBe(60); expect(state.revision).toBe(60);
+1
View File
@@ -389,6 +389,7 @@ describe("GStack 2 managed runtime installer", () => {
const cli = path.join(result.path, "runtime", "cli.js"); const cli = path.join(result.path, "runtime", "cli.js");
const originalMode = (await fs.stat(cli)).mode & 0o777; const originalMode = (await fs.stat(cli)).mode & 0o777;
await fs.chmod(cli, originalMode === 0o600 ? 0o644 : 0o600); await fs.chmod(cli, originalMode === 0o600 ? 0o644 : 0o600);
await expect(validateRuntimeBundle(result.path, { version: "2.0.0", platform: "win32" })).resolves.toBe(true);
await expect(validateRuntimeBundle(result.path, { version: "2.0.0" })).rejects.toMatchObject({ await expect(validateRuntimeBundle(result.path, { version: "2.0.0" })).rejects.toMatchObject({
code: "INSTALL_VALIDATION_FAILED", code: "INSTALL_VALIDATION_FAILED",
}); });
+4 -4
View File
@@ -164,7 +164,7 @@ describe("managed-home destructive boundary", () => {
describe("one config authority", () => { describe("one config authority", () => {
test("compatibility helper and runtime config share config.json", async () => { test("compatibility helper and runtime config share config.json", async () => {
const home = path.join(await root(), "state"); const home = path.join(await root(), "state");
const run = (args: string[]) => spawnSync(configBin, args, { const run = (args: string[]) => spawnSync(process.execPath, [configBin, ...args], {
encoding: "utf8", encoding: "utf8",
env: { ...process.env, GSTACK_HOME: home }, env: { ...process.env, GSTACK_HOME: home },
}); });
@@ -201,20 +201,20 @@ describe("one config authority", () => {
const home = path.join(await root(), "legacy"); const home = path.join(await root(), "legacy");
await fs.mkdir(home); await fs.mkdir(home);
await fs.writeFile(path.join(home, "config.yaml"), "telemetry: community\n"); await fs.writeFile(path.join(home, "config.yaml"), "telemetry: community\n");
const get = spawnSync(configBin, ["get", "telemetry"], { const get = spawnSync(process.execPath, [configBin, "get", "telemetry"], {
encoding: "utf8", encoding: "utf8",
env: { ...process.env, GSTACK_HOME: home }, env: { ...process.env, GSTACK_HOME: home },
}); });
expect(get.status).toBe(0); expect(get.status).toBe(0);
expect(get.stdout).toBe("community"); expect(get.stdout).toBe("community");
const set = spawnSync(configBin, ["set", "telemetry", "off"], { const set = spawnSync(process.execPath, [configBin, "set", "telemetry", "off"], {
encoding: "utf8", encoding: "utf8",
env: { ...process.env, GSTACK_HOME: home }, env: { ...process.env, GSTACK_HOME: home },
}); });
expect(set.status).toBe(0); expect(set.status).toBe(0);
expect(await fs.readFile(path.join(home, "config.yaml"), "utf8")).toBe("telemetry: community\n"); expect(await fs.readFile(path.join(home, "config.yaml"), "utf8")).toBe("telemetry: community\n");
expect(JSON.parse(await fs.readFile(path.join(home, "config.json"), "utf8")).telemetry).toBe("off"); expect(JSON.parse(await fs.readFile(path.join(home, "config.json"), "utf8")).telemetry).toBe("off");
const reread = spawnSync(configBin, ["get", "telemetry"], { const reread = spawnSync(process.execPath, [configBin, "get", "telemetry"], {
encoding: "utf8", encoding: "utf8",
env: { ...process.env, GSTACK_HOME: home }, env: { ...process.env, GSTACK_HOME: home },
}); });
+5 -3
View File
@@ -42,8 +42,9 @@ async function fixture(label = "gstack workflow state ", initialize = true) {
} }
afterEach(async () => { afterEach(async () => {
await Promise.all(temporaryRoots.splice(0).map((root) => for (const root of temporaryRoots.splice(0)) {
fs.rm(root, { recursive: true, force: true }))); await fs.rm(root, { recursive: true, force: true, maxRetries: 5, retryDelay: 50 });
}
}); });
describe("GStack 2 authoritative workflow state", () => { describe("GStack 2 authoritative workflow state", () => {
@@ -150,7 +151,7 @@ describe("GStack 2 authoritative workflow state", () => {
test("all workflow mutations are locked and concurrent evidence writes are not lost", async () => { test("all workflow mutations are locked and concurrent evidence writes are not lost", async () => {
const { home, identity } = await fixture(); const { home, identity } = await fixture();
await beginRun(home, identity.projectId, "qa", { runId: "run_concurrent" }); await beginRun(home, identity.projectId, "qa", { runId: "run_concurrent" });
await Promise.all(Array.from({ length: 24 }, (_, index) => const updates = await Promise.allSettled(Array.from({ length: 24 }, (_, index) =>
updateRunWorkflow(home, identity.projectId, "run_concurrent", { updateRunWorkflow(home, identity.projectId, "run_concurrent", {
addEvidenceProvenance: { addEvidenceProvenance: {
source: "local-test", source: "local-test",
@@ -158,6 +159,7 @@ describe("GStack 2 authoritative workflow state", () => {
capturedAt: `2026-07-16T10:${String(index).padStart(2, "0")}:00.000Z`, capturedAt: `2026-07-16T10:${String(index).padStart(2, "0")}:00.000Z`,
}, },
}))); })));
expect(updates.filter((result) => result.status === "rejected")).toEqual([]);
const inspected = await inspectRun(home, identity.projectId, "run_concurrent"); const inspected = await inspectRun(home, identity.projectId, "run_concurrent");
expect(inspected.reconstruction.evidenceProvenance).toHaveLength(24); expect(inspected.reconstruction.evidenceProvenance).toHaveLength(24);
expect(new Set(inspected.reconstruction.evidenceProvenance.map((entry: any) => entry.reference)).size).toBe(24); expect(new Set(inspected.reconstruction.evidenceProvenance.map((entry: any) => entry.reference)).size).toBe(24);