File size: 4,575 Bytes
e249c6d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 | import { deserialize } from "node:v8";
import type {
FsSafeCopyRead,
FsSafeCopyReply,
FsSafeCopyWrite,
} from "../../infra/fs-safe-copy-worker-contract.js";
import { runtimeProcessEntrypoints } from "../../infra/runtime-process-entrypoints.js";
import {
resolveRuntimeWorkerArgv,
resolveRuntimeWorkerUrl,
} from "../../infra/runtime-worker-url.js";
import { WorkerTaskError, WorkerTaskPool } from "../../infra/worker-task-pool.js";
import { runCommandBuffersWithTimeout } from "../../process/exec-runner.js";
import { resolveGlobalSingleton } from "../../shared/global-singleton.js";
import type { WorktreeFilesystemOptions } from "./filesystem-backend.types.js";
import { WORKTREE_CHECKOUT_TIMEOUT_MS } from "./git.js";
type ReadRuntime = {
pool?: WorkerTaskPool<FsSafeCopyRead, FsSafeCopyReply>;
closing?: Promise<void>;
};
function workerUrl(): URL {
return resolveRuntimeWorkerUrl(runtimeProcessEntrypoints.fsSafeCopy);
}
function assertActive(options: WorktreeFilesystemOptions): void {
options.signal?.throwIfAborted();
options.commitGuard();
}
function checkReply(reply: FsSafeCopyReply): void {
if (reply.type === "failed") {
throw Object.assign(new Error(reply.message), { code: reply.code });
}
}
async function read(
command: FsSafeCopyRead,
options: WorktreeFilesystemOptions,
): Promise<FsSafeCopyReply> {
assertActive(options);
const runtime = resolveGlobalSingleton<ReadRuntime>(
Symbol.for("openclaw.worktreeFilesystemReads"),
() => ({}),
(owned) => {
owned.closing ??= (owned.pool?.close() ?? Promise.resolve()).finally(() => {
owned.pool = undefined;
owned.closing = undefined;
});
return owned.closing;
},
);
if (runtime.closing) {
throw new WorkerTaskError(
"Native worktree reads are closing; retry the operation",
"unavailable",
);
}
const pool = (runtime.pool ??= new WorkerTaskPool<FsSafeCopyRead, FsSafeCopyReply>({
workerUrl: workerUrl(),
maxWorkers: 1,
idleTimeoutMs: 30_000,
}));
const reply = await pool.run(command, {
inputBytes:
command.type === "probe"
? Buffer.byteLength(command.parent)
: command.paths.reduce((bytes, pathname) => bytes + Buffer.byteLength(pathname), 0),
signal: options.signal,
timeoutMs: WORKTREE_CHECKOUT_TIMEOUT_MS,
});
assertActive(options);
checkReply(reply);
return reply;
}
async function write(command: FsSafeCopyWrite, options: WorktreeFilesystemOptions): Promise<void> {
assertActive(options);
// A process owns its descriptor table. Never terminate a JS worker while
// fs-safe's native task still borrows descriptors opened by that worker.
const result = await runCommandBuffersWithTimeout(
[process.execPath, ...resolveRuntimeWorkerArgv(workerUrl())],
{
input: JSON.stringify(command),
beforeInput: () => assertActive(options),
signal: options.signal,
timeoutMs: WORKTREE_CHECKOUT_TIMEOUT_MS,
killGraceMs: WORKTREE_CHECKOUT_TIMEOUT_MS,
killProcessTree: true,
requireProcessTreeExtinction: true,
maxOutputBytes: 64 * 1024,
},
);
assertActive(options);
if (result.code !== 0 || result.termination !== "exit") {
throw new Error(`Native worktree ${command.type} failed: ${result.termination}`);
}
// SAFETY: The private child returns a serialized reply only after its native call settles.
const reply = deserialize(result.stdout) as FsSafeCopyReply;
checkReply(reply);
if (reply.type !== "written") {
throw new Error("Native worktree operation returned no completion receipt");
}
}
export const nativeWorktreeFilesystem = {
async probe(this: void, parent: string, options: WorktreeFilesystemOptions) {
const reply = await read({ type: "probe", parent }, options);
if (reply.type !== "probe") {
throw new Error("Native worktree probe returned an invalid reply");
}
return reply.backend;
},
async readMetadata(this: void, paths: string[], options: WorktreeFilesystemOptions) {
const reply = await read({ type: "metadata", paths }, options);
if (reply.type !== "metadata") {
throw new Error("Native worktree metadata returned an invalid reply");
}
return reply.entries;
},
createSource(this: void, destination: string, options: WorktreeFilesystemOptions) {
return write({ type: "create", destination }, options);
},
copy(this: void, source: string, destination: string, options: WorktreeFilesystemOptions) {
return write({ type: "copy", source, destination }, options);
},
};
|