fix(artifacts): sync nested publication and retire admitted file state

This commit is contained in:
morluto committed 2026-10-24 02:35:22 +08:00
1 parent ddb3987bb4
commit e2eacf6f29
3 files changed
+149 -28

No files matched your search

+7
View File
@@ -116,6 +116,13 @@ artifact can execute successfully.
## Own the lifecycle
`SafeOutputTree` publishes only after file readback, nested-directory fsync
from children to parents, and root-parent fsync on hosts that support directory
syncing. Windows skips directory fsync. Failed writes retain a separate pending
identity admission only until identity is captured; rollback retries retain
unremoved file/directory identities and open handles independently, so confirmed
unlink progress cannot be re-admitted through a closed descriptor.
For tools that acquire resources, identify the owner and the observations that
establish each transition. Include temporary files and one-shot commands, as
well as persistent engine sessions. Use the existing provider contracts;
+25 -28
View File
@@ -66,7 +66,8 @@ export class SafeOutputTree {
readonly #nestedDirectories = new Map<string, DirectoryIdentity>();
readonly #unknownDirectories = new Set<string>();
readonly #ownedFiles = new Map<string, FileIdentity>();
readonly #openFiles = new Map<string, OwnedFileHandle>();
/** Exclusive files awaiting identity admission; admitted identity and handles have separate owners. */
readonly #pendingFiles = new Map<string, OwnedFileHandle>();
readonly #openHandles = new Set<OwnedFileHandle>();
#published = false;
#createdRoot = true;
@@ -211,13 +212,14 @@ export class SafeOutputTree {
);
});
const ownedHandle = this.#trackHandle(handle);
this.#openFiles.set(destination, ownedHandle);
this.#pendingFiles.set(destination, ownedHandle);
const fileIdentity = await readFileIdentity(
ownedHandle.handle,
destination,
);
await assertFilePathIdentity(destination, fileIdentity);
this.#ownedFiles.set(destination, fileIdentity);
this.#pendingFiles.delete(destination);
if (this.platform !== "win32") {
await ownedHandle.handle.chmod(0o600);
if (((await ownedHandle.handle.stat()).mode & 0o777) !== 0o600)
@@ -255,7 +257,7 @@ export class SafeOutputTree {
await this.#assertLineage(lineage);
await assertFilePathIdentity(destination, fileIdentity);
await ownedHandle.handle.sync();
await this.#closeFileHandle(destination, ownedHandle);
await this.#closeHandle(ownedHandle);
const readbackHandle = this.#trackHandle(
await open(destination, constants.O_RDONLY | constants.O_NOFOLLOW),
);
@@ -284,7 +286,7 @@ export class SafeOutputTree {
}
}
/** Sync the owned output tree and prevent further writes through this instance. */
/** Sync files' containing directories from children to root, then publish the root in its parent. */
async commit(signal?: AbortSignal): Promise<void> {
this.#assertWritable();
abortIfNeeded(signal);
@@ -302,7 +304,7 @@ export class SafeOutputTree {
constants.O_RDONLY | constants.O_DIRECTORY,
),
);
let output: OwnedFileHandle | undefined;
const handles: OwnedFileHandle[] = [parent];
let sync:
| { readonly kind: "synced" }
| {
@@ -310,18 +312,21 @@ export class SafeOutputTree {
readonly cause: unknown;
} = { kind: "synced" };
try {
output = this.#trackHandle(
await open(
this.#outputRoot,
constants.O_RDONLY | constants.O_DIRECTORY,
),
);
await assertHandleIdentity(
output.handle,
this.#requireRootIdentity(),
this.#outputRoot,
);
await output.handle.sync();
// A directory's entries become durable before its parent publishes the
// child directory name. Keep only one nested descriptor open at a time.
const directories = [...this.#nestedDirectories].reverse();
directories.push([this.#outputRoot, this.#requireRootIdentity()]);
for (const [path, identity] of directories) {
abortIfNeeded(signal);
const directory = this.#trackHandle(
await open(path, constants.O_RDONLY | constants.O_DIRECTORY),
);
handles.push(directory);
await assertHandleIdentity(directory.handle, identity, path);
await directory.handle.sync();
await this.#closeHandle(directory);
handles.pop();
}
await parent.handle.sync();
await this.#assertExactEntries();
abortIfNeeded(signal);
@@ -330,9 +335,7 @@ export class SafeOutputTree {
sync = { kind: "failed", cause };
}
const closeResults = await Promise.allSettled(
[parent, output]
.filter(isDefined)
.map((handle) => this.#closeHandle(handle)),
handles.map((handle) => this.#closeHandle(handle)),
);
const closeFailures = closeResults
.filter(isRejected)
@@ -609,25 +612,19 @@ export class SafeOutputTree {
throw new AggregateError(failures, "Output tree handles remain open");
}
async #closeFileHandle(path: string, owner: OwnedFileHandle): Promise<void> {
await this.#closeHandle(owner);
this.#openFiles.delete(path);
}
async #admitPendingFiles(): Promise<void> {
for (const [path, owner] of this.#openFiles) {
for (const [path, owner] of this.#pendingFiles) {
let identity = this.#ownedFiles.get(path);
if (identity === undefined) {
identity = await readFileIdentity(owner.handle, path);
await assertFilePathIdentity(path, identity);
this.#ownedFiles.set(path, identity);
}
this.#pendingFiles.delete(path);
}
}
}
const isDefined = <T>(value: T | undefined): value is T => value !== undefined;
const isRejected = (
result: PromiseSettledResult<void>,
): result is PromiseRejectedResult => result.status === "rejected";
@@ -0,0 +1,117 @@
import { createHash } from "node:crypto";
import { access, chmod, readFile, readdir, rm } from "node:fs/promises";
import { dirname, join } from "node:path";
import { Readable } from "node:stream";
import { expect, it, vi } from "vitest";
import { createTestTempDirectory } from "../../fixtures/temporaryDirectory.js";
const observation = vi.hoisted(() => ({
synced: [] as string[],
refuseSync: undefined as string | undefined,
}));
vi.mock("node:fs/promises", async (importOriginal) => {
const filesystem = await importOriginal<typeof import("node:fs/promises")>();
return {
...filesystem,
open: async (...args: Parameters<typeof filesystem.open>) => {
const handle = await filesystem.open(...args);
const sync = handle.sync.bind(handle);
handle.sync = async () => {
const path = String(args[0]);
if (path === observation.refuseSync)
throw new Error("controlled nested directory sync failure");
await sync();
observation.synced.push(path);
};
return handle;
},
};
});
import { SafeOutputTree } from "../../../src/artifacts/SafeOutputTree.js";
it.skipIf(process.platform === "win32")(
"syncs nested directory entries before their parents and refuses publication when one cannot sync",
async () => {
const root = await createTestTempDirectory("rea-tree-directory-sync-");
const bytes = Buffer.from("published bytes");
for (const fail of [false, true]) {
const output = join(root, fail ? "failed" : "success");
const tree = await SafeOutputTree.create(output);
try {
await tree.write("a/b/payload.txt", Readable.from([bytes]), {
bytes: bytes.length,
sha256: createHash("sha256").update(bytes).digest("hex"),
});
observation.synced.length = 0;
observation.refuseSync = fail ? join(output, "a") : undefined;
if (fail) {
await expect(tree.commit()).rejects.toMatchObject({
message: "Could not durably sync extraction output",
cause: { message: "controlled nested directory sync failure" },
});
expect(tree.published).toBe(false);
expect(observation.synced).toEqual([join(output, "a", "b")]);
observation.refuseSync = undefined;
expect(await tree.rollback()).toEqual({
status: "complete",
residualPaths: [],
});
await expect(access(output)).rejects.toMatchObject({
code: "ENOENT",
});
} else {
await tree.commit();
expect(tree.published).toBe(true);
expect(observation.synced).toEqual([
join(output, "a", "b"),
join(output, "a"),
output,
dirname(output),
]);
expect(await readFile(join(output, "a", "b", "payload.txt"))).toEqual(
bytes,
);
}
} finally {
observation.refuseSync = undefined;
await tree.closeHandles();
}
}
},
);
it
.skipIf(process.platform === "win32" || process.getuid?.() === 0)
.each([false, true])(
"resumes rollback after root removal denial without readmitting a closed file (failed write: %s)",
async (failedWrite) => {
const root = await createTestTempDirectory("rea-tree-removal-retry-");
const output = join(root, "output");
const tree = await SafeOutputTree.create(output);
const bytes = Buffer.from("x");
try {
const write = tree.write("partial.txt", Readable.from([bytes]), {
bytes: failedWrite ? 2 : 1,
sha256: createHash("sha256").update(bytes).digest("hex"),
});
if (failedWrite)
await expect(write).rejects.toMatchObject({ reason: "integrity" });
else await write;
await chmod(root, 0o500);
await expect(tree.rollback()).rejects.toMatchObject({
cause: { code: "EACCES", syscall: "rmdir" },
});
expect(await readdir(output)).toEqual([]);
await chmod(root, 0o700);
expect(await tree.rollback()).toEqual({
status: "complete",
residualPaths: [],
});
await expect(access(output)).rejects.toMatchObject({ code: "ENOENT" });
} finally {
await chmod(root, 0o700);
await tree.closeHandles();
await rm(root, { recursive: true, force: true });
}
},
);