From 3bfe82180e1d6a0585518bc44bbf0fead8b98152 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sat, 8 Aug 2026 23:28:51 -0700 Subject: [PATCH] =?UTF-8?q?fix(gateway):=20unbreak=20cloud=20session=20cre?= =?UTF-8?q?ation=20=E2=80=94=20stale=20chat=20metadata,=20unbounded=20tunn?= =?UTF-8?q?el=20hangs,=20swallowed=20dispatch=20errors=20(#120926)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(gateway): rebuild chat metadata when auth-profile snapshots change chat.metadata cached a prepared generation built before the runtime auth-profile store snapshot was published (empty-store fallback) and generationFactsMatch never compared auth state, so the Control UI showed "No models available" after a gateway restart until an unrelated config edit. Capture per-agent auth snapshot revisions in the generation facts, subscribe the metadata lifecycle to auth-store mutations, and run one awaited revision-aware catch-up refresh after listener registration so publications that precede registration are still observed. * fix(gateway): bound cloud worker tunnel startup and surface dispatch failure detail Live stress-testing the Control UI cloud flow found sessions.dispatch hanging unbounded (observed 12+ min) when the worker SSH tunnel could not connect: the runner's exited promise settled only on "close" (a spawn error never settled it), the reconnect loop swallowed every failure with no logging, the poisoned ready promise was re-handed to every later dispatch, and dispatch errors dropped the actionable reason recorded in worker_environments.last_error. - settle exited on the real "exit" event, wire the owner abort signal into spawn, bound stop()'s post-SIGKILL wait, and fail stop honestly when termination is unconfirmed instead of fabricating an exit - add a 60s per-attempt readiness deadline and log each failed connect attempt (bounded, redacted); keep an unconfirmed child tracked and wait for its real exit before retrying - add a 3-minute epoch-fenced startTunnel deadline with a typed, actionable error; detach its cleanup so the deadline holds - append the bounded recorded reason to the five dispatch-visible worker environment error messages (docs already promise these) - remote socket setup: drop "--" from chmod (BSD/macOS chmod treats it as a filename), which blocked every tunnel to a macOS worker host - workspace quiescence: tolerate EPERM without crashing the protocol while keeping unsignalable freeze targets counted as live so quiescence fails closed * test(gateway): type worker child kill mock --- src/agents/auth-profiles.ts | 1 + src/agents/auth-profiles/store.ts | 2 + src/gateway/server-chat-metadata-lifecycle.ts | 27 ++- .../chat-metadata-runtime.test.ts | 66 ++++++- .../server-methods/chat-metadata-runtime.ts | 6 + .../worker-environments/service.test.ts | 42 +++++ src/gateway/worker-environments/service.ts | 36 +++- .../tunnel-ssh-runner.test.ts | 169 ++++++++++++++++++ .../worker-environments/tunnel-ssh-runner.ts | 69 ++++++- .../worker-environments/tunnel.test.ts | 21 +++ src/gateway/worker-environments/tunnel.ts | 46 ++++- .../workspace-quiescence-scripts.ts | 27 ++- 12 files changed, 482 insertions(+), 30 deletions(-) create mode 100644 src/gateway/worker-environments/tunnel-ssh-runner.test.ts diff --git a/src/agents/auth-profiles.ts b/src/agents/auth-profiles.ts index 1275f364eb2a..bbe8dbbf564a 100644 --- a/src/agents/auth-profiles.ts +++ b/src/agents/auth-profiles.ts @@ -63,6 +63,7 @@ export { ensureAuthProfileStoreWithoutExternalProfiles, getPreparedRuntimeAuthProfileStoreSnapshot, getRuntimeAuthProfileStoreSnapshot, + getRuntimeAuthProfileStoreSnapshotRevision, hasAuthProfileStoreSourceForProvider, hasAnyAuthProfileStoreSource, hasLocalAuthProfileStoreSource, diff --git a/src/agents/auth-profiles/store.ts b/src/agents/auth-profiles/store.ts index d8fdc2cf7109..fe3dd1d86bb0 100644 --- a/src/agents/auth-profiles/store.ts +++ b/src/agents/auth-profiles/store.ts @@ -1326,6 +1326,8 @@ export function getPreparedRuntimeAuthProfileStoreSnapshot( return getPreparedRuntimeAuthProfileStoreSnapshotImpl(agentDir, inheritedAuthDir); } +export { getRuntimeAuthProfileStoreSnapshotRevision }; + /** Replace runtime auth-profile snapshots, used by tests and prepared runtimes. */ export function replaceRuntimeAuthProfileStoreSnapshots( entries: Array<{ agentDir?: string; store: AuthProfileStore }>, diff --git a/src/gateway/server-chat-metadata-lifecycle.ts b/src/gateway/server-chat-metadata-lifecycle.ts index f9ac373bb31b..9da5a63de570 100644 --- a/src/gateway/server-chat-metadata-lifecycle.ts +++ b/src/gateway/server-chat-metadata-lifecycle.ts @@ -46,11 +46,15 @@ export async function createGatewayChatMetadataLifecycle(params: { if (params.minimalTestGateway) { return undefined; } - const [{ registerPreparedModelRuntimePublicationListener }, { registerSkillsChangeListener }] = - await Promise.all([ - import("../agents/prepared-model-runtime.js"), - import("../skills/runtime/refresh.js"), - ]); + const [ + { registerRuntimeAuthProfileStoreMutationListener }, + { registerPreparedModelRuntimePublicationListener }, + { registerSkillsChangeListener }, + ] = await Promise.all([ + import("../agents/auth-profiles/runtime-snapshots.js"), + import("../agents/prepared-model-runtime.js"), + import("../skills/runtime/refresh.js"), + ]); const unregisterPreparedModelRuntimePublication = registerPreparedModelRuntimePublicationListener((event) => { if (event.phase === "invalidated") { @@ -67,8 +71,14 @@ export async function createGatewayChatMetadataLifecycle(params: { runtime.invalidate(); refreshLogged(); }); + const unregisterRuntimeAuthProfileStoreMutation = + registerRuntimeAuthProfileStoreMutationListener(() => { + runtime.invalidate(); + refreshLogged(); + }); return { stop: async () => { + unregisterRuntimeAuthProfileStoreMutation(); unregisterPreparedModelRuntimePublication(); unregisterSkillsChange(); }, @@ -85,6 +95,13 @@ export async function createGatewayChatMetadataLifecycle(params: { if (sidecar) { sidecars.push(sidecar); } + // Auth/model snapshots published before listener registration are otherwise never + // observed; one awaited catch-up refresh reconciles facts (revision-aware, no-op when + // fresh) so post-attach reads cannot serve a pre-publication generation. Failures are + // logged, not thrown: startup must not die on a metadata build error. + await runtime.refresh().catch((error: unknown) => { + params.log.warn(`chat metadata catch-up refresh failed: ${String(error)}`); + }); }, read: runtime.read, readStartup: runtime.readStartup, diff --git a/src/gateway/server-methods/chat-metadata-runtime.test.ts b/src/gateway/server-methods/chat-metadata-runtime.test.ts index b8c8d373bf82..4a3b0082fe43 100644 --- a/src/gateway/server-methods/chat-metadata-runtime.test.ts +++ b/src/gateway/server-methods/chat-metadata-runtime.test.ts @@ -37,16 +37,22 @@ function createHarness( let owner = createOwner(config, "first"); let skillsVersion = 1; let pluginRegistryVersion = 1; - const authStore: AuthProfileStore = { version: 1, profiles: {} }; + let authStore: AuthProfileStore | undefined = { version: 1, profiles: {} }; + let authStoreRevision = 1; const getPreparedOwner = vi.fn(() => owner); const getPreparedAuthStore = vi.fn(() => authStore); + const getAuthStoreRevision = vi.fn(() => authStoreRevision); const getSkillsVersion = vi.fn(() => skillsVersion); const getPluginRegistryVersion = vi.fn(() => pluginRegistryVersion); const buildCommands = vi.fn(async () => ({ commands: [{ name: `command-${skillsVersion}-${pluginRegistryVersion}` }], })); const buildProjection = vi.fn( - async ({ facts }: { facts: { owner: PreparedModelRuntimeSnapshot } }) => ({ + async ({ + facts, + }: { + facts: { authStore: AuthProfileStore; owner: PreparedModelRuntimeSnapshot }; + }) => ({ modelCatalog: facts.owner.modelCatalog.entries, models: facts.owner.modelCatalog.entries, }), @@ -67,6 +73,7 @@ function createHarness( deps: { getPreparedOwner, getPreparedAuthStore, + getAuthStoreRevision, getSkillsVersion, getPluginRegistryVersion, buildCommands, @@ -77,6 +84,7 @@ function createHarness( buildCommands, buildProjection, getPluginRegistryVersion, + getAuthStoreRevision, getPreparedAuthStore, getPreparedOwner, getSkillsVersion, @@ -84,6 +92,12 @@ function createHarness( setConfig(next: OpenClawConfig) { config = next; }, + setAuthStore(next: AuthProfileStore | undefined) { + authStore = next; + }, + setAuthStoreRevision(next: number) { + authStoreRevision = next; + }, setOwner(next: PreparedModelRuntimeSnapshot) { owner = next; }, @@ -130,6 +144,7 @@ describe("gateway chat metadata runtime", () => { await harness.runtime.refresh(); harness.getPreparedOwner.mockClear(); harness.getPreparedAuthStore.mockClear(); + harness.getAuthStoreRevision.mockClear(); harness.getSkillsVersion.mockClear(); harness.getPluginRegistryVersion.mockClear(); @@ -139,6 +154,7 @@ describe("gateway chat metadata runtime", () => { expect(first).toBe(second); expect(harness.getPreparedOwner).not.toHaveBeenCalled(); expect(harness.getPreparedAuthStore).not.toHaveBeenCalled(); + expect(harness.getAuthStoreRevision).not.toHaveBeenCalled(); expect(harness.getSkillsVersion).not.toHaveBeenCalled(); expect(harness.getPluginRegistryVersion).not.toHaveBeenCalled(); }); @@ -148,6 +164,7 @@ describe("gateway chat metadata runtime", () => { await harness.runtime.refresh(); harness.getPreparedOwner.mockClear(); harness.getPreparedAuthStore.mockClear(); + harness.getAuthStoreRevision.mockClear(); harness.getSkillsVersion.mockClear(); harness.getPluginRegistryVersion.mockClear(); @@ -169,6 +186,7 @@ describe("gateway chat metadata runtime", () => { expect(harness.buildProjection).toHaveBeenCalledTimes(1); expect(harness.getPreparedOwner).not.toHaveBeenCalled(); expect(harness.getPreparedAuthStore).not.toHaveBeenCalled(); + expect(harness.getAuthStoreRevision).not.toHaveBeenCalled(); expect(harness.getSkillsVersion).not.toHaveBeenCalled(); expect(harness.getPluginRegistryVersion).not.toHaveBeenCalled(); }); @@ -322,6 +340,50 @@ describe("gateway chat metadata runtime", () => { expect(harness.buildProjection).toHaveBeenCalledTimes(1); }); + test("rebuilds after an auth store publishes a newer revision", async () => { + const harness = createHarness(); + harness.setAuthStore(undefined); + harness.buildProjection.mockImplementation(async ({ facts }) => ({ + modelCatalog: facts.owner.modelCatalog.entries, + models: facts.owner.modelCatalog.entries.map((model) => ({ + ...model, + available: Object.keys(facts.authStore.profiles).length > 0, + })), + })); + + await harness.runtime.refresh(); + await expect(harness.runtime.read({ agentId: "main" })).resolves.toMatchObject({ + models: [expect.objectContaining({ available: false })], + }); + + harness.setAuthStore({ + version: 1, + profiles: { "test:default": { type: "api_key", provider: "test" } }, + }); + harness.setAuthStoreRevision(2); + await harness.runtime.refresh(); + + await expect(harness.runtime.read({ agentId: "main" })).resolves.toMatchObject({ + models: [expect.objectContaining({ available: true })], + }); + expect(harness.buildProjection).toHaveBeenCalledTimes(2); + }); + + test("retains a generation while auth store revisions are unchanged", async () => { + const harness = createHarness(); + harness.getPreparedAuthStore.mockImplementation(() => ({ version: 1, profiles: {} })); + await harness.runtime.refresh(); + const first = await harness.runtime.read({ agentId: "main" }); + + await harness.runtime.refresh(); + const second = await harness.runtime.read({ agentId: "main" }); + + expect(second).toBe(first); + expect(harness.buildProjection).toHaveBeenCalledTimes(1); + expect(harness.getAuthStoreRevision).toHaveBeenCalledWith("/tmp/first/agent"); + expect(harness.getAuthStoreRevision).toHaveBeenCalledWith(undefined); + }); + test("refreshes config, catalog-auth, skills, and plugin generations", async () => { const harness = createHarness(); await harness.runtime.refresh(); diff --git a/src/gateway/server-methods/chat-metadata-runtime.ts b/src/gateway/server-methods/chat-metadata-runtime.ts index 150510c0fb76..461db10c1885 100644 --- a/src/gateway/server-methods/chat-metadata-runtime.ts +++ b/src/gateway/server-methods/chat-metadata-runtime.ts @@ -5,6 +5,7 @@ import { } from "../../agents/agent-scope.js"; import { getPreparedRuntimeAuthProfileStoreSnapshot, + getRuntimeAuthProfileStoreSnapshotRevision, type AuthProfileStore, } from "../../agents/auth-profiles.js"; import type { ModelCatalogEntry } from "../../agents/model-catalog.types.js"; @@ -40,6 +41,7 @@ type PreparedAgentFacts = { agentId: string; owner: PreparedModelRuntimeSnapshot; authStore: AuthProfileStore; + authStoreRevision: string; skillsVersion: number; }; @@ -85,6 +87,7 @@ type ChatMetadataRuntimeDeps = { agentDir?: string, inheritedAuthDir?: string, ) => AuthProfileStore | undefined; + getAuthStoreRevision: (agentDir?: string) => number; getSkillsVersion: (workspaceDir?: string) => number; getPluginRegistryVersion: () => number; buildCommands: (params: { @@ -150,6 +153,7 @@ function captureGenerationFacts(deps: ChatMetadataRuntimeDeps): PreparedGenerati version: 1, profiles: {}, }, + authStoreRevision: `${deps.getAuthStoreRevision(owner.agentDir)}:${deps.getAuthStoreRevision(owner.inheritedAuthDir)}`, skillsVersion: deps.getSkillsVersion(workspaceDir), }; }); @@ -177,6 +181,7 @@ function generationFactsMatch( return ( candidate?.agentId === agent.agentId && candidate.owner === agent.owner && + candidate.authStoreRevision === agent.authStoreRevision && candidate.skillsVersion === agent.skillsVersion ); }); @@ -277,6 +282,7 @@ export function createGatewayChatMetadataRuntime(params: { getContext: params.getContext, getPreparedOwner: getPreparedModelCatalogOwnerSnapshot, getPreparedAuthStore: getPreparedRuntimeAuthProfileStoreSnapshot, + getAuthStoreRevision: getRuntimeAuthProfileStoreSnapshotRevision, getSkillsVersion: getSkillsSnapshotVersion, getPluginRegistryVersion: getActivePluginRegistryVersion, buildCommands: defaultBuildCommands, diff --git a/src/gateway/worker-environments/service.test.ts b/src/gateway/worker-environments/service.test.ts index 1702f4681f1f..23e5ed39a44a 100644 --- a/src/gateway/worker-environments/service.test.ts +++ b/src/gateway/worker-environments/service.test.ts @@ -1164,6 +1164,7 @@ describe("worker environment service", () => { workerService.create("development", "request-preparation-failure"), ).rejects.toMatchObject({ code: "bootstrap_failure", + message: expect.stringContaining("npm install requires a released gateway package"), } satisfies Partial); expect(provision).not.toHaveBeenCalled(); @@ -1589,6 +1590,7 @@ describe("worker environment service", () => { await expect(workerService.create("development", "request-malformed")).rejects.toMatchObject({ code: "provider_failure", + message: expect.stringContaining(error), } satisfies Partial); expect(store.list()[0]).toMatchObject({ state: "provisioning", @@ -1621,6 +1623,7 @@ describe("worker environment service", () => { await expect(workerService.create("development", "request-invalid")).rejects.toMatchObject({ code: "invalid_profile", + message: expect.stringContaining("region is required"), } satisfies Partial); const record = expectDefined(store.list()[0], "store.list()[0] test invariant"); expect(record).toMatchObject({ state: "failed", lastError: "region is required" }); @@ -2258,6 +2261,45 @@ describe("worker environment service", () => { expect(order).toEqual(["tunnel-stop", "provider-destroy"]); }); + it("stops a poisoned tunnel start and returns a typed deadline error", async () => { + vi.useFakeTimers(); + seedReady("worker-tunnel-timeout"); + let signalStarted!: () => void; + const started = new Promise((resolve) => { + signalStarted = resolve; + }); + let rejectStart!: (error: Error) => void; + const pendingStart = new Promise((_resolve, reject) => { + rejectStart = reject; + }); + const tunnelManager = { + status: () => "connecting" as const, + start: vi.fn(() => { + signalStarted(); + return pendingStart; + }), + stop: vi.fn(async () => { + rejectStart(new Error("tunnel stopped")); + }), + stopAll: vi.fn(async () => {}), + } as unknown as WorkerTunnelManager; + const workerService = createService(createProvider(), { tunnelManager }); + + const starting = workerService.startTunnel({ + environmentId: "worker-tunnel-timeout", + ownerEpoch: 1, + }); + const rejected = expect(starting).rejects.toMatchObject({ + code: "provider_failure", + message: expect.stringContaining("did not connect within 3 minutes"), + } satisfies Partial); + await started; + await vi.advanceTimersByTimeAsync(3 * 60_000); + + await rejected; + expect(tunnelManager.stop).toHaveBeenCalledWith("worker-tunnel-timeout", 1); + }); + it("adopts an unpersisted provision result before destroying", async () => { const intent = store.createIntent({ environmentId: "worker-pending-destroy", diff --git a/src/gateway/worker-environments/service.ts b/src/gateway/worker-environments/service.ts index 10a9c036d841..eaa071502c06 100644 --- a/src/gateway/worker-environments/service.ts +++ b/src/gateway/worker-environments/service.ts @@ -92,6 +92,8 @@ class WorkerEnvironmentServiceError extends Error { const serviceError = (code: WorkerEnvironmentServiceErrorCode, message: string) => new WorkerEnvironmentServiceError(code, message); const ORPHANED_LEASE_ERROR = "Worker provider no longer recognizes the lease"; +// One poisoned SSH attempt must not hold every later dispatch on the same owner epoch forever. +const TUNNEL_START_TIMEOUT_MS = 3 * 60_000; function workerEnvironmentIdempotencyDigest(idempotencyKey: string): string { return createHash("sha256").update(idempotencyKey).digest("hex"); @@ -614,15 +616,16 @@ export function createWorkerEnvironmentService(options: WorkerEnvironmentService ), ); } catch (error) { + const detail = boundedError(error); if ( error instanceof WorkerProviderError || (error instanceof WorkerEnvironmentServiceError && error.code === "invalid_profile") ) { - move(record, "failed", { lastError: boundedError(error) }); - throw serviceError("invalid_profile", "Worker provider rejected profile"); + move(record, "failed", { lastError: detail }); + throw serviceError("invalid_profile", `Worker provider rejected profile: ${detail}`); } saveError(record, error); - throw serviceError("provider_failure", "Worker provider operation failed"); + throw serviceError("provider_failure", `Worker provider operation failed: ${detail}`); } // A timeout can happen after allocation; retain the same operation id for safe replay. const patch = { leaseId: lease.leaseId, sshEndpoint: lease.ssh }; @@ -654,8 +657,12 @@ export function createWorkerEnvironmentService(options: WorkerEnvironmentService // must happen first because the previous response may have been lost after allocation. installation = await prepareInstallation(record); } catch (error) { - move(record, "failed", { lastError: boundedError(error) }); - throw serviceError("bootstrap_failure", "Worker installation preparation failed"); + const detail = boundedError(error); + move(record, "failed", { lastError: detail }); + throw serviceError( + "bootstrap_failure", + `Worker installation preparation failed: ${detail}`, + ); } } const provisioning = record.state === "requested" ? move(record, "provisioning") : record; @@ -1101,7 +1108,24 @@ export function createWorkerEnvironmentService(options: WorkerEnvironmentService if (!startup) { throw serviceError("invalid_state", "Worker tunnel failed to start"); } - return await startup; + const timeoutError = serviceError( + "provider_failure", + "Worker tunnel did not connect within 3 minutes; check worker SSH reachability and retry", + ); + try { + return await withTimeout(startup, TUNNEL_START_TIMEOUT_MS, { + createError: () => timeoutError, + }); + } catch (error) { + if (error !== timeoutError) { + throw error; + } + // Stop can itself block on an unkillable SSH child; detach it (rejection observed, + // entry stays manager-tracked) so the deadline error is returned on time. Epoch-fenced + // so a stale timed-out attempt can never tear down a newer owner's tunnel. + void tunnels.stop(request.environmentId, request.ownerEpoch).catch(() => undefined); + throw timeoutError; + } }; const stopTunnel = async (environmentId: string, ownerEpoch?: number): Promise => { diff --git a/src/gateway/worker-environments/tunnel-ssh-runner.test.ts b/src/gateway/worker-environments/tunnel-ssh-runner.test.ts new file mode 100644 index 000000000000..f5dc5248483f --- /dev/null +++ b/src/gateway/worker-environments/tunnel-ssh-runner.test.ts @@ -0,0 +1,169 @@ +import type { SpawnOptions } from "node:child_process"; +import { EventEmitter } from "node:events"; +import { PassThrough } from "node:stream"; +import { afterEach, beforeEach, describe, expect, it, type Mock, vi } from "vitest"; + +const spawnMock = vi.hoisted(() => vi.fn()); + +vi.mock("node:child_process", () => ({ spawn: spawnMock })); + +import { createWorkerSshRunner } from "./tunnel-ssh-runner.js"; + +// Returned as the fake-typed union (not ChildProcessWithoutNullStreams) so `child.kill` +// stays a plain vi.fn property; casting to the real type makes kill an unbound method for lint. +function createChild() { + const child = new EventEmitter() as EventEmitter & { + stdin: PassThrough; + stdout: PassThrough; + stderr: PassThrough; + kill: Mock<(signal?: NodeJS.Signals | number) => boolean>; + }; + child.stdin = new PassThrough(); + child.stdout = new PassThrough(); + child.stderr = new PassThrough(); + child.kill = vi.fn(() => true); + return child; +} + +describe("worker SSH process runner", () => { + beforeEach(() => { + spawnMock.mockReset(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it("settles readiness and exit when spawn emits an error without close", async () => { + const child = createChild(); + spawnMock.mockReturnValue(child); + const process = createWorkerSshRunner().start(["missing-ssh"], { timeoutMs: 10_000 }); + + child.emit("error", new Error("spawn failed")); + + await expect(process.ready).rejects.toThrow("Worker SSH tunnel failed"); + await expect(process.exited).resolves.toEqual({ code: null, signal: null }); + }); + + it("settles with the real exit when close lags after SIGKILL", async () => { + vi.useFakeTimers(); + const child = createChild(); + spawnMock.mockReturnValue(child); + const process = createWorkerSshRunner().start(["ssh"], { timeoutMs: 10_000 }); + + const stopping = process.stop(); + expect(child.kill).toHaveBeenCalledWith("SIGTERM"); + await vi.advanceTimersByTimeAsync(1_500); + expect(child.kill).toHaveBeenCalledWith("SIGKILL"); + child.emit("exit", null, "SIGKILL"); + await vi.advanceTimersByTimeAsync(2_000); + + await expect(stopping).resolves.toBeUndefined(); + await expect(process.exited).resolves.toEqual({ code: null, signal: "SIGKILL" }); + await expect(process.ready).rejects.toThrow("Worker SSH tunnel failed"); + }); + + it("fails stop when a SIGKILLed child never reports exit", async () => { + vi.useFakeTimers(); + const child = createChild(); + spawnMock.mockReturnValue(child); + const process = createWorkerSshRunner().start(["ssh"], { timeoutMs: 10_000 }); + + const stopping = process.stop(); + const rejection = expect(stopping).rejects.toThrow("did not exit after SIGKILL"); + await vi.advanceTimersByTimeAsync(1_500); + await vi.advanceTimersByTimeAsync(2_000); + await rejection; + }); + + it("treats a post-exit kill failure as terminal when close is delayed", async () => { + vi.useFakeTimers(); + const child = createChild(); + child.kill = vi.fn(() => false); + spawnMock.mockReturnValue(child); + const process = createWorkerSshRunner().start(["ssh"], { timeoutMs: 10_000 }); + + child.emit("exit", 255, null); + const stopping = process.stop(); + await vi.advanceTimersByTimeAsync(1_500); + await vi.advanceTimersByTimeAsync(2_000); + + await expect(stopping).resolves.toBeUndefined(); + await expect(process.exited).resolves.toEqual({ code: 255, signal: null }); + await expect(process.ready).rejects.toThrow("Worker SSH tunnel failed"); + }); + + it("propagates a stop failure when SIGKILL delivery fails", async () => { + vi.useFakeTimers(); + const child = createChild(); + child.kill = vi.fn(() => false); + spawnMock.mockReturnValue(child); + const process = createWorkerSshRunner().start(["ssh"], { timeoutMs: 10_000 }); + + const stopping = process.stop(); + const rejection = expect(stopping).rejects.toThrow("SIGKILL delivery failed"); + await vi.advanceTimersByTimeAsync(1_500); + await vi.advanceTimersByTimeAsync(2_000); + await rejection; + + let exitedSettled = false; + void process.exited.then(() => { + exitedSettled = true; + }); + await Promise.resolve(); + await Promise.resolve(); + expect(exitedSettled).toBe(false); + }); + + it("keeps exited pending when a live child emits error without close", async () => { + const child = createChild(); + (child as { pid?: number }).pid = 4242; + spawnMock.mockReturnValue(child); + const process = createWorkerSshRunner().start(["ssh"], { timeoutMs: 10_000 }); + + child.emit("error", new Error("kill delivery failed")); + + await expect(process.ready).rejects.toThrow("Worker SSH tunnel failed"); + let exitedSettled = false; + void process.exited.then(() => { + exitedSettled = true; + }); + await Promise.resolve(); + await Promise.resolve(); + expect(exitedSettled).toBe(false); + + child.emit("close", 0, null); + await expect(process.exited).resolves.toEqual({ code: 0, signal: null }); + }); + + it("passes the owner signal to spawn so abort terminates and settles the child", async () => { + const child = createChild(); + spawnMock.mockImplementation((_command: string, _args: string[], options: SpawnOptions) => { + options.signal?.addEventListener( + "abort", + () => { + child.kill("SIGTERM"); + child.emit("error", Object.assign(new Error("aborted"), { name: "AbortError" })); + }, + { once: true }, + ); + return child; + }); + const controller = new AbortController(); + const process = createWorkerSshRunner().start(["ssh"], { + timeoutMs: 10_000, + signal: controller.signal, + }); + + controller.abort(); + + expect(child.kill).toHaveBeenCalledWith("SIGTERM"); + await expect(process.ready).rejects.toThrow("Worker SSH tunnel failed"); + await expect(process.exited).resolves.toEqual({ code: null, signal: null }); + expect(spawnMock).toHaveBeenCalledWith( + "ssh", + [], + expect.objectContaining({ signal: controller.signal }), + ); + }); +}); diff --git a/src/gateway/worker-environments/tunnel-ssh-runner.ts b/src/gateway/worker-environments/tunnel-ssh-runner.ts index 4517d4dffa9f..8a593d0e15bb 100644 --- a/src/gateway/worker-environments/tunnel-ssh-runner.ts +++ b/src/gateway/worker-environments/tunnel-ssh-runner.ts @@ -10,6 +10,7 @@ import { export const WORKER_TUNNEL_READY_MARKER = "OPENCLAW_WORKER_TUNNEL_READY"; const STOP_GRACE_MS = 1_500; +const STOP_KILL_WAIT_MS = 2_000; const STDERR_LIMIT = 4_096; type WorkerSshProcessExit = { @@ -44,10 +45,12 @@ export function createWorkerSshRunner(): WorkerSshRunner { } const child = spawn(command, args, { env: options.baseEnv, + signal: options.signal, stdio: ["pipe", "pipe", "pipe"], windowsHide: true, }); let closed = false; + let exitedSettled = false; let readySettled = false; let resolveReady!: () => void; let rejectReady!: (error: Error) => void; @@ -56,6 +59,9 @@ export function createWorkerSshRunner(): WorkerSshRunner { resolveReady = resolve; rejectReady = reject; }); + // Readiness can reject after its awaiter timed out and moved on (stop()/late close); + // observe it here so lifecycle settles never become unhandled rejections. + void ready.catch(() => {}); const exited = new Promise((resolve) => { resolveExited = resolve; }); @@ -68,6 +74,13 @@ export function createWorkerSshRunner(): WorkerSshRunner { readySettled = true; rejectReady(workerSshProcessError(stderr)); }; + const settleExited = (exit: WorkerSshProcessExit) => { + if (exitedSettled) { + return; + } + exitedSettled = true; + resolveExited(exit); + }; child.stdout.setEncoding("utf8"); child.stdout.on("error", () => {}); child.stdout.on("data", (chunk: string) => { @@ -85,11 +98,34 @@ export function createWorkerSshRunner(): WorkerSshRunner { child.stderr.on("data", (chunk: string) => { stderr = sliceUtf16Safe(`${stderr}${chunk}`, -STDERR_LIMIT); }); - child.once("error", settleReadyError); + child.once("error", () => { + settleReadyError(); + // "error" also fires for abort/kill-delivery failures on a live child; only a child + // that never spawned (no pid) gets a synthesized exit, otherwise close/stop() settle it. + // The no-pid case is terminal: mark it closed so stop() never signals an unspawned child. + if (child.pid === undefined) { + closed = true; + settleExited({ code: null, signal: null }); + } + }); + // "exit" fires before "close", and "close" can be delayed indefinitely while a + // descendant holds a piped stdio descriptor; settle on the real exit so connected + // tunnels awaiting `exited` observe termination without depending on stream closure. + let exitEventResult: WorkerSshProcessExit | undefined; + child.once("exit", (code, signal) => { + exitEventResult = { code, signal }; + settleReadyError(); + settleExited(exitEventResult); + // Release our pipe ends so a descendant holding the other side cannot pin local + // descriptors across retries; this also lets "close" fire promptly. + child.stdin.destroy(); + child.stdout.destroy(); + child.stderr.destroy(); + }); child.once("close", (code, signal) => { closed = true; settleReadyError(); - resolveExited({ code, signal }); + settleExited({ code, signal }); }); child.stdin.on("error", () => {}); if (options.input !== undefined) { @@ -117,9 +153,32 @@ export function createWorkerSshRunner(): WorkerSshRunner { }), ]); clearTimeout(timer); - if (!closed) { - child.kill("SIGKILL"); - await exited; + if (!closed && !exitedSettled) { + // A false return can also mean the child died a moment ago with its "exit" + // event still queued; always take the bounded wait before judging. + const killDelivered = child.kill("SIGKILL"); + let killTimer: ReturnType | undefined; + let killWaitExpired = false; + await Promise.race([ + exited, + new Promise((resolve) => { + killTimer = setTimeout(() => { + killWaitExpired = true; + resolve(); + }, STOP_KILL_WAIT_MS); + killTimer.unref?.(); + }), + ]); + clearTimeout(killTimer); + if (killWaitExpired) { + // Neither delivered SIGKILL nor failed delivery proves termination without + // an exit event; fail the stop so the owner keeps tracking the live child. + throw workerSshProcessError( + killDelivered + ? "SSH child did not exit after SIGKILL; it may still be running" + : "SIGKILL delivery failed; SSH child may still be running", + ); + } } })()); }, diff --git a/src/gateway/worker-environments/tunnel.test.ts b/src/gateway/worker-environments/tunnel.test.ts index ff92a96a825d..2b3dfbc247cd 100644 --- a/src/gateway/worker-environments/tunnel.test.ts +++ b/src/gateway/worker-environments/tunnel.test.ts @@ -93,6 +93,27 @@ describe("worker tunnel manager", () => { await handle.stop(); }); + it("times out a marker-less SSH child and retries", async () => { + vi.useFakeTimers(); + const fake = fakeRunner(); + const manager = createWorkerTunnelManager({ runner: fake.runner, sleep: async () => {} }); + const starting = startTestTunnel(manager, "worker:ready-timeout", 1); + const rejected = expect(starting).rejects.toThrow("stopped before connecting"); + + try { + await waitForStarts(fake.starts, 1); + await vi.advanceTimersByTimeAsync(60_000); + await waitForStarts(fake.starts, 2); + + expect(fake.starts[0]?.process.stopCount).toBe(1); + expect(manager.status("worker:ready-timeout")).toBe("reconnecting"); + } finally { + await manager.stop("worker:ready-timeout"); + await rejected; + vi.useRealTimers(); + } + }); + it("reconnects on the next advertised port after SSH transport exit 255", async () => { const fake = fakeRunner(); const manager = createWorkerTunnelManager({ diff --git a/src/gateway/worker-environments/tunnel.ts b/src/gateway/worker-environments/tunnel.ts index d9a536af1f11..f3f38bc29f90 100644 --- a/src/gateway/worker-environments/tunnel.ts +++ b/src/gateway/worker-environments/tunnel.ts @@ -1,8 +1,11 @@ import { RetrySupervisor } from "../../../packages/retry/src/index.js"; import { sleepWithAbort, type BackoffPolicy } from "../../infra/backoff.js"; +import { withTimeout } from "../../infra/fs-safe.js"; +import { createSubsystemLogger } from "../../logging/subsystem.js"; import type { WorkerSshEndpoint } from "../../plugins/types.js"; import type { SpawnResult } from "../../process/exec.js"; import { createDeferred, type Deferred } from "../../shared/deferred.js"; +import { boundedWorkerError } from "./service-validation.js"; import { advanceWorkerSshAfterTransportExit, prepareWorkerSsh, @@ -30,6 +33,9 @@ import { createWorkerWorkspaceActions, stableWorkerPathComponent } from "./works export type { WorkerTunnelHandle } from "./tunnel-contract.js"; const REMOTE_SOCKET_NAME = "gateway.sock"; const REMOTE_SETUP_TIMEOUT_MS = 20_000; +// A live SSH process without the remote marker is not a usable tunnel. Bound each attempt so the +// retry supervisor can move on instead of pinning the environment forever. +const TUNNEL_READY_TIMEOUT_MS = 60_000; const DEFAULT_STABLE_CONNECTION_MS = 30_000; const DEFAULT_BACKOFF: BackoffPolicy = { initialMs: 250, @@ -37,6 +43,7 @@ const DEFAULT_BACKOFF: BackoffPolicy = { factor: 2, jitter: 0, }; +const tunnelLog = createSubsystemLogger("gateway/worker-tunnel"); const REMOTE_SOCKET_SETUP_SCRIPT = String.raw`set -eu directory=$1 @@ -50,7 +57,7 @@ if [ -e "$directory" ] || [ -L "$directory" ]; then else mkdir -- "$directory" fi -chmod 700 -- "$directory" +chmod 700 "$directory" # no "--": BSD/macOS chmod treats it as a filename; path is script-owned and absolute rm -f -- "$socket" `; @@ -281,7 +288,9 @@ export function createWorkerTunnelManager(options: WorkerTunnelManagerOptions = child = connection.process; childPort = connection.port; entry.process = child; - await child.ready; + await withTimeout(child.ready, TUNNEL_READY_TIMEOUT_MS, { + message: "Worker tunnel did not become ready within 60 seconds", + }); if (!isCurrent(entry)) { await child.stop(); return; @@ -306,14 +315,41 @@ export function createWorkerTunnelManager(options: WorkerTunnelManagerOptions = if (now() - connectedAtMs >= stableConnectionMs) { reconnectSupervisor.reset(); } - } catch { + } catch (error) { if (child && childPort !== undefined) { - const exit = await child.exited.catch(() => undefined); + let stopError: unknown; + let stopFailed = false; + const stopping = child.stop().catch((failure: unknown) => { + stopFailed = true; + stopError = failure; + }); + let exit = await Promise.race([ + child.exited.catch(() => undefined), + stopping.then(() => undefined), + ]); + await stopping; + if (stopFailed) { + // A failed stop means the SSH child may still be running. Never drop it from + // tracking and never retry over it — wait for its real exit first, and keep + // that late exit so transport-exit port rotation still advances. + tunnelLog.warn("worker tunnel stop failed; waiting for SSH child exit", { + environmentId: entry.environmentId, + error: boundedWorkerError(stopError), + connectError: boundedWorkerError(error), + }); + exit = (await child.exited.catch(() => undefined)) ?? exit; + } if (exit && entry.prepared) { advanceWorkerSshAfterTransportExit(entry.prepared, childPort, exit); } } - await child?.stop().catch(() => undefined); + if (isCurrent(entry)) { + tunnelLog.warn("worker tunnel connect attempt failed", { + environmentId: entry.environmentId, + attempt: reconnectSupervisor.attempts + 1, + error: boundedWorkerError(error), + }); + } } finally { if (entry.process === child) { entry.process = undefined; diff --git a/src/gateway/worker-environments/workspace-quiescence-scripts.ts b/src/gateway/worker-environments/workspace-quiescence-scripts.ts index 7d900b7e4a9e..b04ae71bbec9 100644 --- a/src/gateway/worker-environments/workspace-quiescence-scripts.ts +++ b/src/gateway/worker-environments/workspace-quiescence-scripts.ts @@ -92,6 +92,10 @@ function persistLease(targetPath, lease, verifyCurrent) { fs.renameSync(temporary, targetPath); }`; +// Signal sites tolerate ESRCH (gone) without aborting the protocol. EPERM (exists but +// unsignalable, e.g. macOS SIP-protected same-uid processes on shared static-ssh dev hosts) +// must not crash cleanup/resume paths, but a freeze target that returns EPERM stays counted +// as live so quiescence fails closed instead of reporting a still-running process as frozen. export const REMOTE_WORKSPACE_QUIESCE_JS = String.raw`const childProcess = require("node:child_process"); const crypto = require("node:crypto"); const fs = require("node:fs"); @@ -123,13 +127,15 @@ function writeLease(expiresAtMs = Date.now() + watchdogTimeoutMs) { expiresAtMs, }); } +// EPERM on SIGCONT implies the target was never ours to freeze: kill permission checks are +// identical for SIGSTOP and SIGCONT, so any process this uid successfully stopped can be resumed. function resumeProcesses(entries) { for (const entry of entries) { if (processIdentity(entry.pid) !== entry.start) continue; try { process.kill(entry.pid, "SIGCONT"); } catch (error) { - if (!error || error.code !== "ESRCH") throw error; + if (!error || (error.code !== "ESRCH" && error.code !== "EPERM")) throw error; } } } @@ -143,7 +149,7 @@ for (const name of orphanNames) { const orphanPath = path.join(leaseDirectory, name); const lease = parseLease(fs.readFileSync(orphanPath, "utf8"), match[1]); if (lease.watchdog !== null && processIdentity(lease.watchdog.pid) === lease.watchdog.start) { - try { process.kill(lease.watchdog.pid, "SIGTERM"); } catch (error) { if (!error || error.code !== "ESRCH") throw error; } + try { process.kill(lease.watchdog.pid, "SIGTERM"); } catch (error) { if (!error || (error.code !== "ESRCH" && error.code !== "EPERM")) throw error; } } resumeProcesses(lease.processes); fs.unlinkSync(orphanPath); @@ -194,6 +200,11 @@ try { } process.kill(pid, "SIGSTOP"); } catch (error) { + if (error && error.code === "EPERM") { + frozen.delete(pid); + writeLease(); + continue; + } if (!error || error.code !== "ESRCH") throw error; } } @@ -210,7 +221,7 @@ try { } } catch (error) { if (processIdentity(watchdog.pid) === watchdogStart) { - try { process.kill(watchdog.pid, "SIGTERM"); } catch (killError) { if (!killError || killError.code !== "ESRCH") throw killError; } + try { process.kill(watchdog.pid, "SIGTERM"); } catch (killError) { if (!killError || (killError.code !== "ESRCH" && killError.code !== "EPERM")) throw killError; } } resumeProcesses([...frozen].map(([pid, start]) => ({ pid, start }))); try { fs.unlinkSync(leasePath); } catch (unlinkError) { if (!unlinkError || unlinkError.code !== "ENOENT") throw unlinkError; } @@ -254,7 +265,7 @@ function watchdogMain(watchedLeasePath, watchedNonce) { typeof entry.start !== "string" || processIdentity(entry.pid) !== entry.start ) continue; - try { process.kill(entry.pid, "SIGCONT"); } catch (error) { if (!error || error.code !== "ESRCH") throw error; } + try { process.kill(entry.pid, "SIGCONT"); } catch (error) { if (!error || (error.code !== "ESRCH" && error.code !== "EPERM")) throw error; } } watchdogFs.unlinkSync(watchedLeasePath); } catch (error) { @@ -345,7 +356,9 @@ if (validationMode === "final") { if (input.expiresAtMs - Date.now() < 2500) refreshLease(frozenEntries); process.kill(pid, "SIGSTOP"); } catch (error) { - if (!error || error.code !== "ESRCH") throw error; + if (!error || (error.code !== "ESRCH" && error.code !== "EPERM")) throw error; + // Fail-closed either way: the candidate scan below runs without the frozen filter, + // so an EPERM-live process re-registers as a candidate and blocks quiescence. frozen.delete(pid); } } @@ -393,11 +406,11 @@ ${REMOTE_QUIESCENCE_PS_JS} ${REMOTE_QUIESCENCE_LEASE_JS} const input = parseLease(raw, nonce); if (input.watchdog !== null && processIdentity(input.watchdog.pid) === input.watchdog.start) { - try { process.kill(input.watchdog.pid, "SIGTERM"); } catch (error) { if (!error || error.code !== "ESRCH") throw error; } + try { process.kill(input.watchdog.pid, "SIGTERM"); } catch (error) { if (!error || (error.code !== "ESRCH" && error.code !== "EPERM")) throw error; } } for (const entry of input.processes) { if (processIdentity(entry.pid) !== entry.start) continue; - try { process.kill(entry.pid, "SIGCONT"); } catch (error) { if (!error || error.code !== "ESRCH") throw error; } + try { process.kill(entry.pid, "SIGCONT"); } catch (error) { if (!error || (error.code !== "ESRCH" && error.code !== "EPERM")) throw error; } } fs.unlinkSync(leasePath); `;