mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-24 19:35:28 -06:00
fix(workers): complete autonomous cloud desktop startup (#126705)
* fix(gateway): admit recovering workers during startup * fix(gateway): admit recovering nodes during startup * fix(crabbox): bind worker desktop to XFCE session * fix(workers): reuse Git base during workspace transfer large clean/stale worktrees were downloading every tracked file after the verified base pack, crossing transfer authority; selectively checkout desired base-index paths, preserving deletions and symlink confinement. * fix(workers): clone reachable stale workspace commits tip-only origin detection forced published ancestor commits through heavyweight Gateway transfer; the existing exact checkout and manifest verification safely own reachability/fallback. * perf(workers): use blobless origin clones * fix(workers): bundle undici in worker deploy artifact
This commit is contained in:
Peter Steinberger
committed by
GitHub
GitHub
parent
8d2d8377a7
commit
3801331d22
@@ -76,36 +76,42 @@ export function createCrabboxWorkerDesktopSetup(leaseId: string, wallpaperBase64
|
||||
'case "$worker_home" in /*) ;; *) echo "Crabbox worker home is invalid" >&2; exit 1 ;; esac',
|
||||
'as_root() { if [ "$worker_uid" -eq 0 ]; then "$@"; else sudo -n -- "$@"; fi; }',
|
||||
...xfceDesktopEnvironment(),
|
||||
'for required_command in xfconf-query xfdesktop xrandr awk curl flock getent pgrep python3; do command -v "$required_command" >/dev/null 2>&1 || { echo "Required Crabbox desktop command is unavailable: $required_command" >&2; exit 1; }; done',
|
||||
"# The live renderer owns XFCE's D-Bus/session lifetime; import only the named values needed to target it.",
|
||||
"bind_xfdesktop_session() {",
|
||||
' mapfile -t renderer_pids < <(pgrep -u "$worker_uid" -x xfdesktop || true)',
|
||||
' [ "${#renderer_pids[@]}" -eq 1 ] || { echo "Expected exactly one worker-owned XFCE desktop renderer; restart crabbox-desktop.service and retry" >&2; exit 1; }',
|
||||
' renderer_pid="${renderer_pids[0]}"',
|
||||
' if ! exec 8<"/proc/$renderer_pid/environ"; then',
|
||||
' echo "XFCE desktop session changed while it was inspected; restart crabbox-desktop.service and retry" >&2',
|
||||
" exit 1",
|
||||
" fi",
|
||||
" renderer_display=",
|
||||
" DBUS_SESSION_BUS_ADDRESS=",
|
||||
" SESSION_MANAGER=",
|
||||
" unset XDG_RUNTIME_DIR",
|
||||
'for required_command in xfconf-query xfdesktop xrandr awk curl flock getent pgrep pkill python3; do command -v "$required_command" >/dev/null 2>&1 || { echo "Required Crabbox desktop command is unavailable: $required_command" >&2; exit 1; }; done',
|
||||
"read_xfce_process_environment() {",
|
||||
' local process_pid="$1"',
|
||||
' exec 8<"/proc/$process_pid/environ" || return 1',
|
||||
" process_display=",
|
||||
" process_dbus=",
|
||||
" process_runtime_dir=",
|
||||
" while IFS= read -r -d '' process_variable; do",
|
||||
' case "$process_variable" in',
|
||||
' DISPLAY=*) renderer_display="${process_variable#*=}" ;;',
|
||||
' DBUS_SESSION_BUS_ADDRESS=*) DBUS_SESSION_BUS_ADDRESS="${process_variable#*=}" ;;',
|
||||
' SESSION_MANAGER=*) SESSION_MANAGER="${process_variable#*=}" ;;',
|
||||
' XDG_RUNTIME_DIR=*) XDG_RUNTIME_DIR="${process_variable#*=}" ;;',
|
||||
' DISPLAY=*) process_display="${process_variable#*=}" ;;',
|
||||
' DBUS_SESSION_BUS_ADDRESS=*) process_dbus="${process_variable#*=}" ;;',
|
||||
' XDG_RUNTIME_DIR=*) process_runtime_dir="${process_variable#*=}" ;;',
|
||||
" esac",
|
||||
" done <&8",
|
||||
" exec 8<&-",
|
||||
' [ "$renderer_display" = ":99" ] || { echo "XFCE desktop renderer does not use DISPLAY=:99; restart crabbox-desktop.service and retry" >&2; exit 1; }',
|
||||
' [ -n "$DBUS_SESSION_BUS_ADDRESS" ] && [ -n "$SESSION_MANAGER" ] || { echo "XFCE desktop renderer is missing its D-Bus or session manager binding; restart crabbox-desktop.service and retry" >&2; exit 1; }',
|
||||
' case "${XDG_RUNTIME_DIR:-}" in ""|/*) ;; *) echo "XFCE desktop renderer has an invalid XDG_RUNTIME_DIR" >&2; exit 1 ;; esac',
|
||||
"}",
|
||||
"bind_xfdesktop_session",
|
||||
"export DBUS_SESSION_BUS_ADDRESS SESSION_MANAGER",
|
||||
'[ -z "${XDG_RUNTIME_DIR:-}" ] || export XDG_RUNTIME_DIR',
|
||||
"# XFCE owns the D-Bus session; the image's original renderer may have been launched outside it.",
|
||||
'mapfile -t session_pids < <(pgrep -u "$worker_uid" -x xfce4-session || true)',
|
||||
'[ "${#session_pids[@]}" -eq 1 ] || { echo "Expected exactly one worker-owned XFCE session; restart crabbox-desktop.service and retry" >&2; exit 1; }',
|
||||
'session_pid="${session_pids[0]}"',
|
||||
'read_xfce_process_environment "$session_pid" || { echo "XFCE session changed while it was inspected; restart crabbox-desktop.service and retry" >&2; exit 1; }',
|
||||
'[ "$process_display" = ":99" ] || { echo "XFCE session does not use DISPLAY=:99; restart crabbox-desktop.service and retry" >&2; exit 1; }',
|
||||
'DBUS_SESSION_BUS_ADDRESS="$process_dbus"',
|
||||
"unset XDG_RUNTIME_DIR",
|
||||
'XDG_RUNTIME_DIR="$process_runtime_dir"',
|
||||
'[ -n "$DBUS_SESSION_BUS_ADDRESS" ] || { echo "XFCE session is missing its D-Bus binding; restart crabbox-desktop.service and retry" >&2; exit 1; }',
|
||||
'case "$XDG_RUNTIME_DIR" in ""|/*) ;; *) echo "XFCE session has an invalid XDG_RUNTIME_DIR" >&2; exit 1 ;; esac',
|
||||
"export DBUS_SESSION_BUS_ADDRESS",
|
||||
'[ -z "$XDG_RUNTIME_DIR" ] || export XDG_RUNTIME_DIR',
|
||||
"bind_xfdesktop_renderer() {",
|
||||
' mapfile -t renderer_pids < <(pgrep -u "$worker_uid" -x xfdesktop || true)',
|
||||
' [ "${#renderer_pids[@]}" -eq 1 ] || return 1',
|
||||
' renderer_pid="${renderer_pids[0]}"',
|
||||
' read_xfce_process_environment "$renderer_pid" || return 1',
|
||||
' [ "$process_display" = "$DISPLAY" ] && [ "$process_dbus" = "$DBUS_SESSION_BUS_ADDRESS" ]',
|
||||
"}",
|
||||
"setup_dir=$(mktemp -d)",
|
||||
"trap 'rm -rf -- \"$setup_dir\"' EXIT",
|
||||
...heredoc("browser", "WORKER_BROWSER_LAUNCHER_EOF", browserLauncher(leaseId)),
|
||||
@@ -120,6 +126,13 @@ export function createCrabboxWorkerDesktopSetup(leaseId: string, wallpaperBase64
|
||||
'as_root install -d -o "$worker_user" -g "$worker_group" -m 0755 "$worker_home/.local" "$worker_home/.local/share" "$worker_home/.local/share/backgrounds"',
|
||||
'wallpaper_path="$worker_home/.local/share/backgrounds/openclaw-worker.png"',
|
||||
'as_root install -o "$worker_user" -g "$worker_group" -m 0644 "$setup_dir/wallpaper.png" "$wallpaper_path"',
|
||||
"# Setup precedes node enrollment, so re-home only this worker's renderer before publishing it.",
|
||||
'pkill -TERM -u "$worker_uid" -x xfdesktop || true',
|
||||
'for _attempt in $(seq 1 20); do pgrep -u "$worker_uid" -x xfdesktop >/dev/null || break; sleep 0.1; done',
|
||||
'pkill -KILL -u "$worker_uid" -x xfdesktop || true',
|
||||
'nohup xfdesktop >"$worker_home/.cache/openclaw/xfdesktop.log" 2>&1 </dev/null &',
|
||||
"for _attempt in $(seq 1 40); do bind_xfdesktop_renderer && break; sleep 0.1; done",
|
||||
'bind_xfdesktop_renderer || { echo "XFCE desktop renderer did not converge on the worker session" >&2; exit 1; }',
|
||||
"mapfile -t backdrop_roots < <(",
|
||||
" {",
|
||||
" xfconf-query -c xfce4-desktop -l | sed -n 's#\\(/backdrop/[^/]*/[^/]*/workspace[^/]*\\)/.*#\\1#p'",
|
||||
@@ -133,7 +146,7 @@ export function createCrabboxWorkerDesktopSetup(leaseId: string, wallpaperBase64
|
||||
"done",
|
||||
'renderer_pid_before_reload="$renderer_pid"',
|
||||
"xfdesktop --reload",
|
||||
"bind_xfdesktop_session",
|
||||
'bind_xfdesktop_renderer || { echo "XFCE desktop renderer lost its worker session during reload" >&2; exit 1; }',
|
||||
'[ "$renderer_pid" = "$renderer_pid_before_reload" ] || { echo "XFCE desktop renderer changed during reload; restart crabbox-desktop.service and retry" >&2; exit 1; }',
|
||||
].join("\n");
|
||||
}
|
||||
|
||||
@@ -423,29 +423,52 @@ describe("Crabbox worker provider", () => {
|
||||
expect(desktopSetupText).not.toContain(". /var/lib/crabbox/desktop.env");
|
||||
expect(desktopSetupText).not.toContain("/var/lib/crabbox/browser.env");
|
||||
expect(desktopSetupLines).not.toContain("export DISPLAY");
|
||||
expect(desktopSetupText).toContain(
|
||||
'mapfile -t session_pids < <(pgrep -u "$worker_uid" -x xfce4-session || true)',
|
||||
);
|
||||
expect(desktopSetupText).toContain("Expected exactly one worker-owned XFCE session");
|
||||
expect(desktopSetupText).toContain('session_pid="${session_pids[0]}"');
|
||||
expect(desktopSetupText).toContain('read_xfce_process_environment "$session_pid"');
|
||||
expect(desktopSetupText).toContain(
|
||||
'mapfile -t renderer_pids < <(pgrep -u "$worker_uid" -x xfdesktop || true)',
|
||||
);
|
||||
expect(desktopSetupText).toContain("Expected exactly one worker-owned XFCE desktop renderer");
|
||||
expect(desktopSetupText).toContain('renderer_pid="${renderer_pids[0]}"');
|
||||
expect(desktopSetupText).toContain('exec 8<"/proc/$renderer_pid/environ"');
|
||||
expect(desktopSetupText).toContain('read_xfce_process_environment "$renderer_pid"');
|
||||
expect(desktopSetupText).toContain('exec 8<"/proc/$process_pid/environ"');
|
||||
for (const [name, target] of [
|
||||
["DISPLAY", "renderer_display"],
|
||||
["DBUS_SESSION_BUS_ADDRESS", "DBUS_SESSION_BUS_ADDRESS"],
|
||||
["SESSION_MANAGER", "SESSION_MANAGER"],
|
||||
["XDG_RUNTIME_DIR", "XDG_RUNTIME_DIR"],
|
||||
["DISPLAY", "process_display"],
|
||||
["DBUS_SESSION_BUS_ADDRESS", "process_dbus"],
|
||||
["XDG_RUNTIME_DIR", "process_runtime_dir"],
|
||||
]) {
|
||||
expect(desktopSetupText).toContain(`${name}=*) ${target}="\${process_variable#*=}"`);
|
||||
}
|
||||
expect(desktopSetupText).toContain('[ "$renderer_display" = ":99" ]');
|
||||
expect(desktopSetupText).toContain('[ "$process_display" = ":99" ]');
|
||||
expect(desktopSetupText).toContain('DBUS_SESSION_BUS_ADDRESS="$process_dbus"');
|
||||
expect(desktopSetupLines).toContain("unset XDG_RUNTIME_DIR");
|
||||
expect(desktopSetupText).toContain('XDG_RUNTIME_DIR="$process_runtime_dir"');
|
||||
expect(desktopSetupText).toContain('[ -n "$DBUS_SESSION_BUS_ADDRESS" ]');
|
||||
expect(desktopSetupText).not.toContain("SESSION_MANAGER");
|
||||
expect(desktopSetupText).toContain(
|
||||
'[ -n "$DBUS_SESSION_BUS_ADDRESS" ] && [ -n "$SESSION_MANAGER" ]',
|
||||
'[ "$process_display" = "$DISPLAY" ] && [ "$process_dbus" = "$DBUS_SESSION_BUS_ADDRESS" ]',
|
||||
);
|
||||
expect(desktopSetupText).toContain(
|
||||
'case "${XDG_RUNTIME_DIR:-}" in ""|/*) ;; *) echo "XFCE desktop renderer has an invalid XDG_RUNTIME_DIR"',
|
||||
'case "$XDG_RUNTIME_DIR" in ""|/*) ;; *) echo "XFCE session has an invalid XDG_RUNTIME_DIR"',
|
||||
);
|
||||
expect(desktopSetupText).toContain("export DBUS_SESSION_BUS_ADDRESS");
|
||||
expect(desktopSetupText).toContain('[ -z "$XDG_RUNTIME_DIR" ] || export XDG_RUNTIME_DIR');
|
||||
for (const signal of ["TERM", "KILL"]) {
|
||||
expect(desktopSetupText).toContain(`pkill -${signal} -u "$worker_uid" -x xfdesktop || true`);
|
||||
}
|
||||
expect(desktopSetupText).toContain('pgrep -u "$worker_uid" -x xfdesktop >/dev/null || break');
|
||||
expect(desktopSetupText).toContain(
|
||||
'nohup xfdesktop >"$worker_home/.cache/openclaw/xfdesktop.log" 2>&1 </dev/null &',
|
||||
);
|
||||
expect(desktopSetupText).toMatch(
|
||||
/for _attempt in \$\(seq 1 \d+\); do bind_xfdesktop_renderer && break; sleep 0\.1; done/u,
|
||||
);
|
||||
expect(desktopSetupText).toContain(
|
||||
"XFCE desktop renderer did not converge on the worker session",
|
||||
);
|
||||
expect(desktopSetupText).toContain("export DBUS_SESSION_BUS_ADDRESS SESSION_MANAGER");
|
||||
expect(desktopSetupText).toContain('[ -z "${XDG_RUNTIME_DIR:-}" ] || export XDG_RUNTIME_DIR');
|
||||
expect(desktopSetupText).not.toMatch(/(?:^|\n)\s*(?:\.|source)\s+[^\n]*\/proc\//u);
|
||||
expect(desktopSetupText).not.toMatch(/(?:^|\n)\s*eval(?:\s|$)/u);
|
||||
expect(desktopSetupText).not.toMatch(/(?:^|\n)\s*(?:\.|source)\s+[^\n]*\.env/u);
|
||||
@@ -470,11 +493,17 @@ describe("Crabbox worker provider", () => {
|
||||
'wallpaper_path="$worker_home/.local/share/backgrounds/openclaw-worker.png"',
|
||||
);
|
||||
expect(desktopSetupText).toContain('for backdrop in "${backdrop_roots[@]}"; do');
|
||||
const sessionExportIndex = desktopSetupText.indexOf(
|
||||
"export DBUS_SESSION_BUS_ADDRESS SESSION_MANAGER",
|
||||
);
|
||||
const sessionExportIndex = desktopSetupText.indexOf("export DBUS_SESSION_BUS_ADDRESS");
|
||||
const sessionExtractionIndex = desktopSetupText.indexOf(
|
||||
'DBUS_SESSION_BUS_ADDRESS=*) DBUS_SESSION_BUS_ADDRESS="${process_variable#*=}"',
|
||||
'read_xfce_process_environment "$session_pid"',
|
||||
);
|
||||
const terminateRendererIndex = desktopSetupText.indexOf(
|
||||
'pkill -TERM -u "$worker_uid" -x xfdesktop',
|
||||
);
|
||||
const killRendererIndex = desktopSetupText.indexOf('pkill -KILL -u "$worker_uid" -x xfdesktop');
|
||||
const launchRendererIndex = desktopSetupText.indexOf("nohup xfdesktop");
|
||||
const convergeRendererIndex = desktopSetupText.indexOf(
|
||||
'bind_xfdesktop_renderer || { echo "XFCE desktop renderer did not converge',
|
||||
);
|
||||
const firstXfconfIndex = desktopSetupText.indexOf("xfconf-query -c xfce4-desktop");
|
||||
const xrandrIndex = desktopSetupText.indexOf("xrandr --listmonitors");
|
||||
@@ -488,16 +517,20 @@ describe("Crabbox worker provider", () => {
|
||||
);
|
||||
expect(sessionExtractionIndex).toBeGreaterThan(-1);
|
||||
expect(sessionExportIndex).toBeGreaterThan(sessionExtractionIndex);
|
||||
expect(sessionExportIndex).toBeGreaterThan(-1);
|
||||
expect(firstXfconfIndex).toBeGreaterThan(sessionExportIndex);
|
||||
expect(terminateRendererIndex).toBeGreaterThan(sessionExportIndex);
|
||||
expect(killRendererIndex).toBeGreaterThan(terminateRendererIndex);
|
||||
expect(launchRendererIndex).toBeGreaterThan(killRendererIndex);
|
||||
expect(convergeRendererIndex).toBeGreaterThan(launchRendererIndex);
|
||||
expect(firstXfconfIndex).toBeGreaterThan(convergeRendererIndex);
|
||||
expect(xrandrIndex).toBeGreaterThan(sessionExportIndex);
|
||||
expect(lastImageIndex).toBeGreaterThan(-1);
|
||||
expect(saveRendererIndex).toBeGreaterThan(lastImageIndex);
|
||||
expect(reloadRendererIndex).toBeGreaterThan(saveRendererIndex);
|
||||
expect(verifyRendererIndex).toBeGreaterThan(reloadRendererIndex);
|
||||
expect(desktopSetupLines.filter((line) => line === "bind_xfdesktop_session")).toHaveLength(2);
|
||||
expect(desktopSetupText).not.toMatch(/pkill[^\n]*xfdesktop/u);
|
||||
expect(desktopSetupText).not.toContain("nohup xfdesktop");
|
||||
expect(desktopSetupText.slice(reloadRendererIndex, verifyRendererIndex)).toContain(
|
||||
"bind_xfdesktop_renderer",
|
||||
);
|
||||
expect(desktopSetupText).not.toMatch(/pkill -(?:TERM|KILL) -x xfdesktop/u);
|
||||
expect(desktopSetupText).not.toContain("def ellipse");
|
||||
expect(desktopSetupText).not.toContain("import struct");
|
||||
expect(desktopSetupText).not.toContain(".svg");
|
||||
|
||||
@@ -11,6 +11,12 @@ const PLAYWRIGHT_BROWSER_REGISTRY_INIT =
|
||||
' registry = new Registry(require(import_path20.default.join(packageRoot, "browsers.json")));';
|
||||
const WORKER_BROWSER_RUNTIME_COMPOSITION = `import { createAttachedBrowserToolRuntime } from "../../extensions/browser/runtime-api.js";
|
||||
export default { createAttachedBrowserToolRuntime };`;
|
||||
const UNDICI_REQUIRE_BOOTSTRAP = [
|
||||
'import { createRequire } from "node:module";',
|
||||
"const requireUndici = createRequire(import.meta.url);\n",
|
||||
'return requireUndici("undici") as typeof import("undici");',
|
||||
] as const;
|
||||
const WORKER_UNDICI_IMPORT = 'import * as bundledUndici from "undici";';
|
||||
|
||||
/** Composes bundled-plugin runtime and removes dependency package reads from the worker build. */
|
||||
export function createWorkerDeployBuildPlugin(rootDir = process.cwd()) {
|
||||
@@ -19,6 +25,9 @@ export function createWorkerDeployBuildPlugin(rootDir = process.cwd()) {
|
||||
const browserRuntimeBridgePath = fs.realpathSync(
|
||||
path.resolve("src/worker/worker-deploy-browser-runtime.ts"),
|
||||
);
|
||||
const undiciDispatcherOptionsPath = fs.realpathSync(
|
||||
path.resolve("src/infra/net/undici-dispatcher-options.ts"),
|
||||
);
|
||||
const packageJson = JSON.parse(
|
||||
fs.readFileSync(path.join(playwrightRoot, "package.json"), "utf8"),
|
||||
) as { name: string; version: string };
|
||||
@@ -46,6 +55,22 @@ export function createWorkerDeployBuildPlugin(rootDir = process.cwd()) {
|
||||
if (resolvedId === browserRuntimeBridgePath) {
|
||||
return WORKER_BROWSER_RUNTIME_COMPOSITION;
|
||||
}
|
||||
if (resolvedId === undiciDispatcherOptionsPath) {
|
||||
if (
|
||||
code.includes(WORKER_UNDICI_IMPORT) &&
|
||||
code.includes("return bundledUndici;") &&
|
||||
UNDICI_REQUIRE_BOOTSTRAP.every((fragment) => !code.includes(fragment))
|
||||
) {
|
||||
return code;
|
||||
}
|
||||
if (UNDICI_REQUIRE_BOOTSTRAP.some((fragment) => !code.includes(fragment))) {
|
||||
this.error("undici dispatcher bootstrap changed; update the worker deploy transform");
|
||||
}
|
||||
return code
|
||||
.replace(UNDICI_REQUIRE_BOOTSTRAP[0], WORKER_UNDICI_IMPORT)
|
||||
.replace(UNDICI_REQUIRE_BOOTSTRAP[1], "")
|
||||
.replace(UNDICI_REQUIRE_BOOTSTRAP[2], "return bundledUndici;");
|
||||
}
|
||||
if (
|
||||
resolvedId !== coreBundlePath ||
|
||||
!id.replaceAll("\\", "/").endsWith("/playwright-core/lib/coreBundle.js")
|
||||
|
||||
@@ -130,10 +130,11 @@ describe("node pairing rate limit", () => {
|
||||
caps: [],
|
||||
commands: [],
|
||||
deviceIdentityPath: identityPath,
|
||||
prePairDevice: false,
|
||||
});
|
||||
|
||||
expect(response.ok).toBe(true);
|
||||
expect(response.payload).toMatchObject({ type: "hello-ok" });
|
||||
expect(response.ok, JSON.stringify(response)).toBe(true);
|
||||
expect(response.payload).toMatchObject({ type: "hello-ok", auth: { role: "node" } });
|
||||
expect(nodeRegistry?.get(identity.deviceId)).toMatchObject({
|
||||
nodeId: identity.deviceId,
|
||||
});
|
||||
|
||||
@@ -165,6 +165,8 @@ export async function admitGatewayConnect(context: GatewayConnectPhaseContext) {
|
||||
|
||||
const isNodeClient = isStartupNodeConnect(connectParams);
|
||||
const startupPending = isStartupPending?.() === true;
|
||||
// Node enrollment is an awaited startup dependency: authenticated node admission
|
||||
// must complete while ordinary methods and other clients remain startup-gated.
|
||||
if (startupPending && !isNodeClient) {
|
||||
await rejectGatewayStartupConnect(context);
|
||||
return undefined;
|
||||
|
||||
@@ -0,0 +1,96 @@
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import { useAutoCleanupTempDirTracker } from "../../../test/helpers/temp-dir.js";
|
||||
import type { SpawnResult } from "../../process/exec.js";
|
||||
import { createNodeWorkerWorkspaceFallback } from "./node-worker-workspace-fallback.js";
|
||||
|
||||
const runCommandWithTimeout = vi.hoisted(() => vi.fn());
|
||||
|
||||
vi.mock("../../process/exec.js", async (importOriginal) => ({
|
||||
...(await importOriginal<typeof import("../../process/exec.js")>()),
|
||||
runCommandWithTimeout,
|
||||
}));
|
||||
|
||||
const COMMIT = "a".repeat(40);
|
||||
const ADVERTISED_TIP = "b".repeat(40);
|
||||
const ORIGIN = "https://example.invalid/openclaw.git";
|
||||
const MANIFEST_REF = `sha256:${"c".repeat(64)}`;
|
||||
const REMOTE_WORKSPACE = "/node/workspace";
|
||||
const tempDirs = useAutoCleanupTempDirTracker(afterEach);
|
||||
|
||||
type WorkspaceExec = Parameters<typeof createNodeWorkerWorkspaceFallback>[0];
|
||||
|
||||
function spawnResult(stdout = "", code = 0): SpawnResult {
|
||||
return { stdout, stderr: "", code, signal: null, killed: false, termination: "exit" };
|
||||
}
|
||||
|
||||
function cleanWorkspace(): string {
|
||||
const root = tempDirs.make("node-worker-origin-workspace-");
|
||||
runCommandWithTimeout.mockReset();
|
||||
runCommandWithTimeout.mockImplementation(async (argv: string[]) => {
|
||||
const args = argv.slice(argv.indexOf("-C") + 2);
|
||||
switch (args.join(" ")) {
|
||||
case "rev-parse --show-toplevel":
|
||||
return spawnResult(root);
|
||||
case "status --porcelain=v1 --untracked-files=all":
|
||||
return spawnResult();
|
||||
case "rev-parse HEAD":
|
||||
return spawnResult(COMMIT);
|
||||
case "remote get-url origin":
|
||||
return spawnResult(ORIGIN);
|
||||
case `ls-remote --heads --tags -- ${ORIGIN}`:
|
||||
return spawnResult(`${ADVERTISED_TIP}\trefs/heads/main\n`);
|
||||
default:
|
||||
throw new Error(`unexpected local Git command: ${args.join(" ")}`);
|
||||
}
|
||||
});
|
||||
return root;
|
||||
}
|
||||
|
||||
describe("node worker workspace origin fallback", () => {
|
||||
it("clones a clean commit without requiring it to be an advertised ref tip", async () => {
|
||||
const localPath = cleanWorkspace();
|
||||
const exec = vi.fn<WorkspaceExec>(async ({ argv }) => ({
|
||||
...spawnResult(argv[0] === "node" ? MANIFEST_REF : ""),
|
||||
workspaceDir: REMOTE_WORKSPACE,
|
||||
}));
|
||||
|
||||
await expect(
|
||||
createNodeWorkerWorkspaceFallback(exec).trySyncWorkspace(
|
||||
{ localPath, sessionId: "session-1", generation: 1 },
|
||||
MANIFEST_REF,
|
||||
),
|
||||
).resolves.toEqual({
|
||||
kind: "synced",
|
||||
result: { mode: "git", remoteWorkspaceDir: REMOTE_WORKSPACE, manifestRef: MANIFEST_REF },
|
||||
});
|
||||
|
||||
expect(exec.mock.calls.map(([command]) => command.argv)).toEqual([
|
||||
expect.arrayContaining(["clone", "--filter=blob:none", ORIGIN]),
|
||||
expect.arrayContaining(["checkout", COMMIT]),
|
||||
expect.arrayContaining(["node", REMOTE_WORKSPACE, COMMIT]),
|
||||
]);
|
||||
expect(runCommandWithTimeout).not.toHaveBeenCalledWith(
|
||||
expect.arrayContaining(["ls-remote"]),
|
||||
expect.anything(),
|
||||
);
|
||||
});
|
||||
|
||||
it.each([
|
||||
{ operation: "clone", reason: "clone-failed", commandCount: 1 },
|
||||
{ operation: "checkout", reason: "checkout-failed", commandCount: 2 },
|
||||
] as const)("preserves the $reason fallback", async ({ operation, reason, commandCount }) => {
|
||||
const localPath = cleanWorkspace();
|
||||
const exec = vi.fn<WorkspaceExec>(async ({ argv }) => ({
|
||||
...spawnResult("", argv.includes(operation) ? 1 : 0),
|
||||
workspaceDir: REMOTE_WORKSPACE,
|
||||
}));
|
||||
|
||||
await expect(
|
||||
createNodeWorkerWorkspaceFallback(exec).trySyncWorkspace(
|
||||
{ localPath, sessionId: "session-1", generation: 1 },
|
||||
MANIFEST_REF,
|
||||
),
|
||||
).resolves.toEqual({ kind: "fallback", reason });
|
||||
expect(exec).toHaveBeenCalledTimes(commandCount);
|
||||
});
|
||||
});
|
||||
@@ -29,7 +29,6 @@ type OriginFallbackReason =
|
||||
| "not-git-workspace"
|
||||
| "not-repository-root"
|
||||
| "origin-unavailable"
|
||||
| "origin-unpublished"
|
||||
| "workspace-dirty"
|
||||
| "workspace-transfer-required";
|
||||
|
||||
@@ -144,17 +143,7 @@ async function inspectEligibleOrigin(localPath: string): Promise<OriginInspectio
|
||||
if (!COMMIT_PATTERN.test(commit) || !origin) {
|
||||
return { kind: "fallback", reason: "origin-unavailable" };
|
||||
}
|
||||
let refs: string;
|
||||
try {
|
||||
refs = await localGit(root, ["ls-remote", "--heads", "--tags", "--", origin]);
|
||||
} catch {
|
||||
return { kind: "fallback", reason: "origin-unavailable" };
|
||||
}
|
||||
return refs
|
||||
.split("\n")
|
||||
.some((line) => line.slice(0, commit.length) === commit && /\srefs\//u.test(line))
|
||||
? { kind: "eligible", identity: { commit, origin, root } }
|
||||
: { kind: "fallback", reason: "origin-unpublished" };
|
||||
return { kind: "eligible", identity: { commit, origin, root } };
|
||||
} catch {
|
||||
return { kind: "fallback", reason: "inspection-failed" };
|
||||
}
|
||||
@@ -183,6 +172,7 @@ export function createNodeWorkerWorkspaceFallback(exec: WorkspaceExec) {
|
||||
"-c",
|
||||
"init.templateDir=",
|
||||
"clone",
|
||||
"--filter=blob:none",
|
||||
"--no-checkout",
|
||||
"--",
|
||||
identity.origin,
|
||||
|
||||
@@ -198,9 +198,10 @@ describe("node worker transfer client", () => {
|
||||
const rawManifest = serializeWorkerWorkspaceManifest({
|
||||
version: 1,
|
||||
baseCommit: null,
|
||||
directories: ["nested"],
|
||||
entries: [
|
||||
{
|
||||
path: "result.txt",
|
||||
path: "nested/result.txt",
|
||||
type: "file",
|
||||
mode: 0o644,
|
||||
size: body.byteLength,
|
||||
@@ -272,6 +273,9 @@ describe("node worker transfer client", () => {
|
||||
transfer: { direction: "download", token: "test-token", manifestRef },
|
||||
}),
|
||||
).resolves.toBe(manifestRef);
|
||||
await expect(
|
||||
fs.readFile(path.join(workspaceDir, "nested", "result.txt"), "utf8"),
|
||||
).resolves.toBe("pinned transfer\n");
|
||||
expect(requestCount).toBe(2);
|
||||
expect(connectionCount).toBe(1);
|
||||
expect(hidPeerCertificate).toBe(true);
|
||||
@@ -621,7 +625,23 @@ describe("node worker transfer client", () => {
|
||||
}
|
||||
});
|
||||
|
||||
it("materializes a Git workspace with argv-only commands", async () => {
|
||||
it.each([
|
||||
{
|
||||
description: "reuses Git-base tracked files without requesting unavailable blobs",
|
||||
changed: false,
|
||||
replaceSymlinkAncestor: false,
|
||||
},
|
||||
{
|
||||
description: "downloads changed and nested files without restoring deleted Git-base paths",
|
||||
changed: true,
|
||||
replaceSymlinkAncestor: false,
|
||||
},
|
||||
{
|
||||
description: "replaces a Git-base symlink ancestor without changing files outside staging",
|
||||
changed: false,
|
||||
replaceSymlinkAncestor: true,
|
||||
},
|
||||
])("$description", async ({ changed, replaceSymlinkAncestor }) => {
|
||||
transferDebug.mockClear();
|
||||
const root = tempDirs.make("node-worker-transfer-git-");
|
||||
const source = path.join(root, "source");
|
||||
@@ -629,9 +649,32 @@ describe("node worker transfer client", () => {
|
||||
await fs.mkdir(source);
|
||||
await git(source, ["init", "--quiet", "--object-format=sha1"]);
|
||||
await fs.writeFile(path.join(source, "tracked.txt"), "tracked from gateway\n");
|
||||
await git(source, ["add", "tracked.txt"]);
|
||||
await fs.writeFile(path.join(source, "script.sh"), "#!/bin/sh\nexit 0\n", { mode: 0o755 });
|
||||
await fs.writeFile(path.join(source, "deleted.txt"), "deleted after commit\n");
|
||||
await fs.symlink("tracked.txt", path.join(source, "tracked-link"));
|
||||
const outsideSentinel = path.join(root, "outside", "file.txt");
|
||||
if (replaceSymlinkAncestor) {
|
||||
await fs.mkdir(path.dirname(outsideSentinel));
|
||||
await fs.writeFile(outsideSentinel, "outside must stay unchanged\n");
|
||||
await fs.symlink("../outside", path.join(source, "nested"));
|
||||
}
|
||||
await git(source, ["add", "."]);
|
||||
await git(source, ["commit", "--quiet", "-m", "base"]);
|
||||
const commit = await git(source, ["rev-parse", "HEAD"]);
|
||||
if (changed) {
|
||||
await fs.writeFile(path.join(source, "tracked.txt"), "changed on gateway\n");
|
||||
await fs.chmod(path.join(source, "tracked.txt"), 0o755);
|
||||
await fs.unlink(path.join(source, "tracked-link"));
|
||||
await fs.symlink("script.sh", path.join(source, "tracked-link"));
|
||||
await fs.unlink(path.join(source, "deleted.txt"));
|
||||
await fs.mkdir(path.join(source, "nested"));
|
||||
await fs.writeFile(path.join(source, "nested", "file.txt"), "new nested content\n");
|
||||
}
|
||||
if (replaceSymlinkAncestor) {
|
||||
await fs.unlink(path.join(source, "nested"));
|
||||
await fs.mkdir(path.join(source, "nested"));
|
||||
await fs.writeFile(path.join(source, "nested", "file.txt"), "safe nested content\n");
|
||||
}
|
||||
const snapshot = await readActualWorkspaceManifest({ root: source, baseCommit: commit });
|
||||
const rawManifest = serializeWorkerWorkspaceManifest(snapshot.manifest);
|
||||
const packed = await runCommandBuffered(
|
||||
@@ -640,11 +683,24 @@ describe("node worker transfer client", () => {
|
||||
);
|
||||
expect(packed.termination, packed.stderr.toString("utf8")).toBe("exit");
|
||||
expect(packed.code).toBe(0);
|
||||
const tracked = snapshot.manifest.entries.find(
|
||||
(entry) => entry.type === "file" && entry.path === "tracked.txt",
|
||||
);
|
||||
if (tracked?.type !== "file") {
|
||||
throw new Error("test Git workspace has no tracked file");
|
||||
}
|
||||
const downloadablePaths = new Set([
|
||||
...(changed ? ["nested/file.txt", "tracked.txt"] : []),
|
||||
...(replaceSymlinkAncestor ? ["nested/file.txt"] : []),
|
||||
]);
|
||||
const filesByHash = new Map(
|
||||
snapshot.manifest.entries.flatMap((entry) =>
|
||||
entry.type === "file" ? [[entry.sha256, path.join(source, entry.path)] as const] : [],
|
||||
entry.type === "file" && downloadablePaths.has(entry.path)
|
||||
? [[entry.sha256, path.join(source, entry.path)] as const]
|
||||
: [],
|
||||
),
|
||||
);
|
||||
const requestedBlobs: string[] = [];
|
||||
const server = createHttpServer((req, res) => {
|
||||
void (async () => {
|
||||
if (req.url?.endsWith("/manifest")) {
|
||||
@@ -658,12 +714,15 @@ describe("node worker transfer client", () => {
|
||||
return;
|
||||
}
|
||||
const sha256 = req.url?.match(/\/blobs\/([a-f0-9]{64})$/u)?.[1];
|
||||
const file = sha256 ? filesByHash.get(sha256) : undefined;
|
||||
if (file) {
|
||||
const body = await fs.readFile(file);
|
||||
res.writeHead(200, { "content-length": String(body.byteLength) });
|
||||
res.end(body);
|
||||
return;
|
||||
if (sha256) {
|
||||
requestedBlobs.push(sha256);
|
||||
const file = filesByHash.get(sha256);
|
||||
if (file) {
|
||||
const body = await fs.readFile(file);
|
||||
res.writeHead(200, { "content-length": String(body.byteLength) });
|
||||
res.end(body);
|
||||
return;
|
||||
}
|
||||
}
|
||||
res.writeHead(404).end();
|
||||
})().catch((error: unknown) => {
|
||||
@@ -686,10 +745,31 @@ describe("node worker transfer client", () => {
|
||||
}),
|
||||
).resolves.toBe(snapshot.manifestRef);
|
||||
await expect(fs.readFile(path.join(workspaceDir, "tracked.txt"), "utf8")).resolves.toBe(
|
||||
"tracked from gateway\n",
|
||||
changed ? "changed on gateway\n" : "tracked from gateway\n",
|
||||
);
|
||||
expect((await fs.stat(path.join(workspaceDir, "tracked.txt"))).mode & 0o777).toBe(
|
||||
changed ? 0o755 : 0o644,
|
||||
);
|
||||
expect((await fs.stat(path.join(workspaceDir, "script.sh"))).mode & 0o777).toBe(0o755);
|
||||
await expect(fs.readlink(path.join(workspaceDir, "tracked-link"))).resolves.toBe(
|
||||
changed ? "script.sh" : "tracked.txt",
|
||||
);
|
||||
expect(requestedBlobs).toEqual([...filesByHash.keys()]);
|
||||
await expect(git(workspaceDir, ["rev-parse", "HEAD"])).resolves.toBe(commit);
|
||||
await expect(git(workspaceDir, ["status", "--porcelain=v1"])).resolves.toBe("");
|
||||
if (changed) {
|
||||
await expect(fs.access(path.join(workspaceDir, "deleted.txt"))).rejects.toMatchObject({
|
||||
code: "ENOENT",
|
||||
});
|
||||
}
|
||||
if (changed || replaceSymlinkAncestor) {
|
||||
expect((await fs.lstat(path.join(workspaceDir, "nested"))).isDirectory()).toBe(true);
|
||||
await expect(
|
||||
fs.readFile(path.join(workspaceDir, "nested", "file.txt"), "utf8"),
|
||||
).resolves.toBe(changed ? "new nested content\n" : "safe nested content\n");
|
||||
}
|
||||
if (!changed && !replaceSymlinkAncestor) {
|
||||
await expect(git(workspaceDir, ["status", "--porcelain=v1"])).resolves.toBe("");
|
||||
}
|
||||
expect(transferDebug).toHaveBeenCalledWith(
|
||||
"node worker workspace transfer completed",
|
||||
expect.objectContaining({
|
||||
@@ -706,6 +786,11 @@ describe("node worker transfer client", () => {
|
||||
await new Promise<void>((resolve) => {
|
||||
server.close(() => resolve());
|
||||
});
|
||||
if (replaceSymlinkAncestor) {
|
||||
await expect(fs.readFile(outsideSentinel, "utf8")).resolves.toBe(
|
||||
"outside must stay unchanged\n",
|
||||
);
|
||||
}
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
@@ -9,7 +9,11 @@ import {
|
||||
MAX_WORKSPACE_MANIFEST_BYTES,
|
||||
MAX_WORKSPACE_INVENTORY_TOTAL_BYTES,
|
||||
} from "../gateway/worker-environments/workspace-inventory-limits.js";
|
||||
import { parseWorkerWorkspaceManifest } from "../gateway/worker-environments/workspace-manifest.js";
|
||||
import {
|
||||
parseWorkerWorkspaceManifest,
|
||||
type WorkerWorkspaceManifestEntry,
|
||||
} from "../gateway/worker-environments/workspace-manifest.js";
|
||||
import { absoluteEntryMatches } from "../gateway/worker-environments/workspace-reconcile-fs.js";
|
||||
import { workerWorkspaceTransferPaths } from "../gateway/worker-environments/workspace-result-staging.js";
|
||||
import { REMOTE_WORKSPACE_MANIFEST_JS } from "../gateway/worker-environments/workspace-sync-scripts.js";
|
||||
import { isPathInside } from "../infra/path-guards.js";
|
||||
@@ -193,6 +197,7 @@ async function initializeGitWorkspace(params: {
|
||||
manifestHome: string;
|
||||
packPath: string;
|
||||
baseCommit: string;
|
||||
entries: WorkerWorkspaceManifestEntry[];
|
||||
signal?: AbortSignal;
|
||||
}): Promise<void> {
|
||||
const objectFormat = params.baseCommit.length === 40 ? "sha1" : "sha256";
|
||||
@@ -234,18 +239,33 @@ async function initializeGitWorkspace(params: {
|
||||
const index = await git(["ls-files", "--stage", "-z"], {
|
||||
maxOutputBytes: MAX_WORKSPACE_MANIFEST_BYTES,
|
||||
});
|
||||
const gitlinks = index
|
||||
.split("\0")
|
||||
.filter(Boolean)
|
||||
.flatMap((record) => {
|
||||
const separator = record.indexOf("\t");
|
||||
return separator >= 0 && record.startsWith("160000 ") ? [record.slice(separator + 1)] : [];
|
||||
});
|
||||
const gitlinks: string[] = [];
|
||||
const basePaths = new Set<string>();
|
||||
for (const record of index.split("\0").filter(Boolean)) {
|
||||
const separator = record.indexOf("\t");
|
||||
if (separator < 0) {
|
||||
continue;
|
||||
}
|
||||
const indexedPath = record.slice(separator + 1);
|
||||
if (record.startsWith("160000 ")) {
|
||||
gitlinks.push(indexedPath);
|
||||
} else {
|
||||
basePaths.add(indexedPath);
|
||||
}
|
||||
}
|
||||
if (gitlinks.length > 0) {
|
||||
await git(["update-index", "--skip-worktree", "-z", "--stdin"], {
|
||||
input: `${gitlinks.join("\0")}\0`,
|
||||
});
|
||||
}
|
||||
const checkoutPaths = params.entries
|
||||
.map((entry) => entry.path)
|
||||
.filter((entryPath) => basePaths.has(entryPath));
|
||||
if (checkoutPaths.length > 0) {
|
||||
await git(["checkout-index", "-z", "--stdin"], {
|
||||
input: `${checkoutPaths.join("\0")}\0`,
|
||||
});
|
||||
}
|
||||
await fsp.rm(params.packPath, { force: true });
|
||||
}
|
||||
|
||||
@@ -391,11 +411,9 @@ async function downloadWorkspace(params: {
|
||||
MAX_WORKSPACE_MANIFEST_BYTES,
|
||||
);
|
||||
const manifest = parseWorkerWorkspaceManifest(raw.toString("utf8"), params.transfer.manifestRef);
|
||||
const parent = path.dirname(params.workspaceDir);
|
||||
const workspaceName = path.basename(params.workspaceDir);
|
||||
const stagingWorkspace = await tempWorkspace({
|
||||
rootDir: parent,
|
||||
prefix: `.${workspaceName}.workspace-transfer-`,
|
||||
rootDir: path.dirname(params.workspaceDir),
|
||||
prefix: `.${path.basename(params.workspaceDir)}.workspace-transfer-`,
|
||||
});
|
||||
const staging = stagingWorkspace.dir;
|
||||
try {
|
||||
@@ -423,6 +441,7 @@ async function downloadWorkspace(params: {
|
||||
manifestHome: params.manifestHome,
|
||||
packPath,
|
||||
baseCommit: manifest.baseCommit,
|
||||
entries: manifest.entries,
|
||||
signal: params.signal,
|
||||
});
|
||||
}
|
||||
@@ -432,6 +451,9 @@ async function downloadWorkspace(params: {
|
||||
}
|
||||
for (const entry of manifest.entries) {
|
||||
const destination = workspacePath(staging, entry.path);
|
||||
if (manifest.baseCommit && (await absoluteEntryMatches(destination, entry))) {
|
||||
continue;
|
||||
}
|
||||
await fsp.mkdir(path.dirname(destination), { recursive: true, mode: 0o700 });
|
||||
await fsp.rm(destination, { recursive: true, force: true });
|
||||
if (entry.type === "symlink") {
|
||||
|
||||
@@ -35,6 +35,36 @@ describe("worker deploy build plugin", () => {
|
||||
expect(transformed).not.toContain("was not composed by the build");
|
||||
});
|
||||
|
||||
it("bundles the undici dispatcher dependency without a worker runtime require", () => {
|
||||
const dispatcherPath = path.resolve("src/infra/net/undici-dispatcher-options.ts");
|
||||
const source = fs.readFileSync(dispatcherPath, "utf8");
|
||||
const plugin = createWorkerDeployBuildPlugin();
|
||||
|
||||
const transformed = plugin.transform.call({ error: fail }, source, dispatcherPath);
|
||||
|
||||
expect(transformed).toContain('import * as bundledUndici from "undici";');
|
||||
expect(transformed).toContain("return bundledUndici;");
|
||||
expect(transformed).toContain('return override as typeof import("undici");');
|
||||
expect(transformed).not.toContain('import { createRequire } from "node:module";');
|
||||
expect(transformed).not.toContain("const requireUndici = createRequire(import.meta.url);");
|
||||
expect(transformed).not.toContain('requireUndici("undici")');
|
||||
expect(plugin.transform.call({ error: fail }, transformed!, dispatcherPath)).toBe(transformed);
|
||||
});
|
||||
|
||||
it("fails closed when the undici dispatcher bootstrap shape changes", () => {
|
||||
const dispatcherPath = path.resolve("src/infra/net/undici-dispatcher-options.ts");
|
||||
const source = fs.readFileSync(dispatcherPath, "utf8");
|
||||
const plugin = createWorkerDeployBuildPlugin();
|
||||
|
||||
expect(() =>
|
||||
plugin.transform.call(
|
||||
{ error: fail },
|
||||
source.replace('return requireUndici("undici")', 'return changedUndici("undici")'),
|
||||
dispatcherPath,
|
||||
),
|
||||
).toThrow("undici dispatcher bootstrap changed");
|
||||
});
|
||||
|
||||
it("inlines Playwright package identity without a runtime manifest read", () => {
|
||||
const coreBundlePath = path.resolve("node_modules/playwright-core/lib/coreBundle.js");
|
||||
const source = fs.readFileSync(coreBundlePath, "utf8");
|
||||
|
||||
Reference in New Issue
Block a user