From 3801331d226b8128b2cd4bce7c2df732bae298b6 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Thu, 20 Aug 2026 08:29:47 -0700 Subject: [PATCH] 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 --- .../src/crabbox-worker-desktop-setup.ts | 63 ++++++---- .../src/crabbox-worker-provider.test.ts | 73 ++++++++---- scripts/lib/worker-deploy-build-plugin.mts | 25 ++++ .../server.node-pairing-rate-limit.test.ts | 5 +- .../server/ws-connection/connect-admission.ts | 2 + .../node-worker-workspace-fallback.test.ts | 96 +++++++++++++++ .../node-worker-workspace-fallback.ts | 14 +-- .../node-worker-transfer-client.test.ts | 109 ++++++++++++++++-- src/node-host/node-worker-transfer-client.ts | 46 ++++++-- .../worker-deploy-build-plugin.test.ts | 30 +++++ 10 files changed, 380 insertions(+), 83 deletions(-) create mode 100644 src/gateway/worker-environments/node-worker-workspace-fallback.test.ts diff --git a/extensions/crabbox/src/crabbox-worker-desktop-setup.ts b/extensions/crabbox/src/crabbox-worker-desktop-setup.ts index 8f7f0081cf55..9cf8cf586d81 100644 --- a/extensions/crabbox/src/crabbox-worker-desktop-setup.ts +++ b/extensions/crabbox/src/crabbox-worker-desktop-setup.ts @@ -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 &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"); } diff --git a/extensions/crabbox/src/crabbox-worker-provider.test.ts b/extensions/crabbox/src/crabbox-worker-provider.test.ts index 53e3af098e10..2a03f0cb03aa 100644 --- a/extensions/crabbox/src/crabbox-worker-provider.test.ts +++ b/extensions/crabbox/src/crabbox-worker-provider.test.ts @@ -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 { '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"); diff --git a/scripts/lib/worker-deploy-build-plugin.mts b/scripts/lib/worker-deploy-build-plugin.mts index 71f43cf91bbd..d1ec6485344e 100644 --- a/scripts/lib/worker-deploy-build-plugin.mts +++ b/scripts/lib/worker-deploy-build-plugin.mts @@ -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") diff --git a/src/gateway/server.node-pairing-rate-limit.test.ts b/src/gateway/server.node-pairing-rate-limit.test.ts index 909604fa7e88..741f96f1ef09 100644 --- a/src/gateway/server.node-pairing-rate-limit.test.ts +++ b/src/gateway/server.node-pairing-rate-limit.test.ts @@ -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, }); diff --git a/src/gateway/server/ws-connection/connect-admission.ts b/src/gateway/server/ws-connection/connect-admission.ts index 875fcdf26818..0d478911f9b3 100644 --- a/src/gateway/server/ws-connection/connect-admission.ts +++ b/src/gateway/server/ws-connection/connect-admission.ts @@ -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; diff --git a/src/gateway/worker-environments/node-worker-workspace-fallback.test.ts b/src/gateway/worker-environments/node-worker-workspace-fallback.test.ts new file mode 100644 index 000000000000..2cb20343f1eb --- /dev/null +++ b/src/gateway/worker-environments/node-worker-workspace-fallback.test.ts @@ -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()), + 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[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(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(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); + }); +}); diff --git a/src/gateway/worker-environments/node-worker-workspace-fallback.ts b/src/gateway/worker-environments/node-worker-workspace-fallback.ts index f63d144b6a6b..43c957f9a670 100644 --- a/src/gateway/worker-environments/node-worker-workspace-fallback.ts +++ b/src/gateway/worker-environments/node-worker-workspace-fallback.ts @@ -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 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, diff --git a/src/node-host/node-worker-transfer-client.test.ts b/src/node-host/node-worker-transfer-client.test.ts index cf15640ba082..5f66cd801741 100644 --- a/src/node-host/node-worker-transfer-client.test.ts +++ b/src/node-host/node-worker-transfer-client.test.ts @@ -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((resolve) => { server.close(() => resolve()); }); + if (replaceSymlinkAncestor) { + await expect(fs.readFile(outsideSentinel, "utf8")).resolves.toBe( + "outside must stay unchanged\n", + ); + } } }); }); diff --git a/src/node-host/node-worker-transfer-client.ts b/src/node-host/node-worker-transfer-client.ts index 112fe9db8d03..d36d70b70d30 100644 --- a/src/node-host/node-worker-transfer-client.ts +++ b/src/node-host/node-worker-transfer-client.ts @@ -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 { 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(); + 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") { diff --git a/test/scripts/worker-deploy-build-plugin.test.ts b/test/scripts/worker-deploy-build-plugin.test.ts index 45b9f0a2dc23..358b3133155c 100644 --- a/test/scripts/worker-deploy-build-plugin.test.ts +++ b/test/scripts/worker-deploy-build-plugin.test.ts @@ -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");