diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index 877f12d8ae1..f927e8224cc 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -2193,6 +2193,18 @@ or `cleanup`) and its timeout in milliseconds. Command output is excluded from that diagnostic. A timeout does not prove the remote process stopped or that a save failed before publication. Reconcile the existing workspace and operation receipts before recovery; the transport does not automatically repeat the command. +The internal helper commands suppress Node runtime warnings so their fixed +failure labels stay separate from warning text. Actual command errors still +fail the operation. This does not change an agent process's environment. + +Loading an encrypted saved snapshot cancels its storage request after five +minutes without incoming bytes, including a wait for response headers. Incoming +bytes renew that idle deadline, so a large download can take longer while it +makes progress. The failure is `remote_workspace_snapshot_download_stalled`; +private download scratch is removed before returning and the saved snapshot is +unchanged. Preparation offers its existing bounded retry only after rechecking +the same clean saved version and workspace owner. Caller cancellation still +cancels the request and does not become a download timeout. Export cleanup keeps this typed timeout diagnostic in the operation's `metadata.exportCleanup.lastError`. Other cleanup errors retain a generic message. The saved snapshot stays acknowledged and the cleanup obligation stays pending diff --git a/server/src/services/remote-workspace-snapshot-storage.test.ts b/server/src/services/remote-workspace-snapshot-storage.test.ts index b34c6888a3c..1451c747793 100644 --- a/server/src/services/remote-workspace-snapshot-storage.test.ts +++ b/server/src/services/remote-workspace-snapshot-storage.test.ts @@ -2,6 +2,9 @@ import { randomUUID } from "node:crypto"; import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; +import { createServer, type ServerResponse } from "node:http"; +import type { AddressInfo } from "node:net"; +import { setTimeout as wait } from "node:timers/promises"; import { Readable } from "node:stream"; import { appendFileSync, readdirSync } from "node:fs"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; @@ -9,6 +12,7 @@ import { eq } from "drizzle-orm"; import { applyPendingMigrations, closeRegisteredClients, createDb, executionWorkspaceSnapshots } from "@paperclipai/db"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js"; import { createLocalDiskStorageProvider } from "../storage/local-disk-provider.js"; +import { createS3StorageProvider } from "../storage/s3-provider.js"; import { createStorageService } from "../storage/service.js"; import type { StorageService } from "../storage/types.js"; import { localEncryptedProvider } from "../secrets/local-encrypted-provider.js"; @@ -341,7 +345,8 @@ suite("remote workspace snapshot storage", () => { destroy(error, callback) { closed = true; callback(error); }, }); storage.getObject = vi.fn(async (_companyId, _objectKey, options) => { - expect(options?.signal).toBe(controller.signal); + expect(options?.signal).toBeInstanceOf(AbortSignal); + expect(options?.signal?.aborted).toBe(false); return { stream: body, contentLength: ciphertext.length }; }); const scratchRoot = path.join(root, "private-scratch"); await fs.mkdir(scratchRoot); @@ -357,4 +362,98 @@ suite("remote workspace snapshot storage", () => { await expect(fs.access(targetPath)).rejects.toMatchObject({ code: "ENOENT" }); } finally { temp.mockRestore(); } }); + it.each(["headers", "body"])("closes a stalled S3 %s request before returning a retryable download failure", async phase => { + const f = await fixture(), ref = await uploadWorkspaceSnapshot(f.upload); + const ciphertext = await fs.readFile(path.join(root, "objects", ref.objectKey)); + let ready!: () => void, closed!: () => void; + const started = new Promise(resolve => { ready = resolve; }); + const disconnected = new Promise(resolve => { closed = resolve; }); + const server = createServer((req, res) => { + req.socket.once("close", closed); + if (phase === "headers") { ready(); return; } + res.writeHead(200, { "content-length": String(ciphertext.length) }); + res.write(ciphertext.subarray(0, 1)); + }); + await new Promise(resolve => server.listen(0, "127.0.0.1", resolve)); + vi.stubEnv("AWS_ACCESS_KEY_ID", "fixture-access-key"); + vi.stubEnv("AWS_SECRET_ACCESS_KEY", "fixture-secret-key"); + vi.stubEnv("AWS_SESSION_TOKEN", ""); + const remote = createStorageService(createS3StorageProvider({ bucket: "bucket", region: "us-east-1", forcePathStyle: true, + endpoint: `http://127.0.0.1:${(server.address() as AddressInfo).port}` })); + const get = remote.getObject.bind(remote); + remote.getObject = async (...args) => { + const object = await get(...args); + object.stream.once("data", ready); + return object; + }; + const controller = new AbortController(), targetPath = path.join(root, "stalled.pcws"); + const scratch = path.join(root, "download-scratch"); await fs.mkdir(scratch); + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); + const result = loadWorkspaceSnapshotArchive({ storage: remote, reference: ref, targetPath, + signal: controller.signal, temporaryParent: scratch }).then(() => ({ succeeded: true }), error => error); + try { + await started; + await vi.advanceTimersByTimeAsync(300_001); + const outcome = await Promise.race([result, wait(100).then(() => ({ pending: true }))]); + expect(outcome).toMatchObject({ name: "TimeoutError", message: "remote_workspace_snapshot_download_stalled" }); + await Promise.race([disconnected, wait(1_000).then(() => { throw new Error("S3 request still connected after timeout"); })]); + expect(controller.signal.aborted).toBe(false); + expect(await fs.readdir(scratch)).toEqual([]); + await expect(fs.access(targetPath)).rejects.toMatchObject({ code: "ENOENT" }); + expect(await getWorkspaceSnapshot(db, { ...f, snapshotId: ref.id })).toEqual(ref); + await loadWorkspaceSnapshotArchive({ storage, reference: ref, targetPath }); + expect(await fs.readFile(targetPath)).toEqual(await fs.readFile(f.archivePath)); + } finally { + controller.abort(); await result; + vi.useRealTimers(); server.closeAllConnections(); + await new Promise(resolve => server.close(() => resolve())); + } + }); + it("lets an S3 download exceed five minutes while bytes keep arriving", async () => { + const f = await fixture(), ref = await uploadWorkspaceSnapshot(f.upload); + const ciphertext = await fs.readFile(path.join(root, "objects", ref.objectKey)); + let response!: ServerResponse, consumed = 0; + const observed = Array.from({ length: 4 }, () => { + let resolve!: () => void; + const promise = new Promise(ready => { resolve = ready; }); + return { promise, resolve }; + }); + const width = Math.ceil(ciphertext.length / 4); + const server = createServer((_req, res) => { + response = res; res.writeHead(200, { "content-length": String(ciphertext.length) }); + res.write(ciphertext.subarray(0, width)); + }); + await new Promise(resolve => server.listen(0, "127.0.0.1", resolve)); + vi.stubEnv("AWS_ACCESS_KEY_ID", "fixture-access-key"); + vi.stubEnv("AWS_SECRET_ACCESS_KEY", "fixture-secret-key"); + vi.stubEnv("AWS_SESSION_TOKEN", ""); + const remote = createStorageService(createS3StorageProvider({ bucket: "bucket", region: "us-east-1", forcePathStyle: true, + endpoint: `http://127.0.0.1:${(server.address() as AddressInfo).port}` })); + const get = remote.getObject.bind(remote); + remote.getObject = async (...args) => { + const object = await get(...args); + object.stream.on("data", () => observed[consumed++]?.resolve()); + return object; + }; + const controller = new AbortController(), targetPath = path.join(root, "progressing.pcws"); + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); + const result = loadWorkspaceSnapshotArchive({ storage: remote, reference: ref, targetPath, signal: controller.signal }); + const settled = result.catch(error => error); + try { + await observed[0]!.promise; + for (let index = 1; index < 4; index += 1) { + await vi.advanceTimersByTimeAsync(240_000); + response.write(ciphertext.subarray(index * width, (index + 1) * width)); + await observed[index]!.promise; + } + response.end(); await result; + expect(await fs.readFile(targetPath)).toEqual(await fs.readFile(f.archivePath)); + await vi.advanceTimersByTimeAsync(300_001); + expect(controller.signal.aborted).toBe(false); + } finally { + controller.abort(); await settled; + vi.useRealTimers(); server.closeAllConnections(); + await new Promise(resolve => server.close(() => resolve())); + } + }); }); diff --git a/server/src/services/remote-workspace-snapshot-storage.ts b/server/src/services/remote-workspace-snapshot-storage.ts index a9bae63c069..697ce5a0821 100644 --- a/server/src/services/remote-workspace-snapshot-storage.ts +++ b/server/src/services/remote-workspace-snapshot-storage.ts @@ -17,6 +17,7 @@ const SCHEMA = "paperclip.workspace-snapshot.v1"; const KEY_SCHEMA = "paperclip.workspace-snapshot-key.v1"; const MAGIC = Buffer.from("PCWSE1\n"); const FORMAT = "paperclip.remote-workspace-archive.v1"; +const DOWNLOAD_IDLE_TIMEOUT_MS = 5 * 60_000; type SnapshotRow = typeof executionWorkspaceSnapshots.$inferSelect; export interface WorkspaceSnapshotRef { schema: typeof SCHEMA; @@ -304,14 +305,25 @@ export async function loadWorkspaceSnapshotArchive(input: { try { input.signal?.throwIfAborted(); const ciphertext = path.join(scratch, "ciphertext"), plaintext = path.join(scratch, "plaintext"); - const object = await input.storage.getObject(ref.companyId, ref.objectKey, { signal: input.signal }); let received = 0; const digest = createHash("sha256"); - const bounded = new Transform({ transform(bytes, _encoding, callback) { - received += bytes.length; digest.update(bytes); - callback(received > ref.byteSize ? new Error("remote_workspace_snapshot_object_size_mismatch") : null, bytes); - } }); - try { await pipeline(object.stream, bounded, createWriteStream(ciphertext, { flags: "wx", mode: 0o600 }), { signal: input.signal }); } - finally { object.stream.destroy(); } + const stalled = new DOMException("remote_workspace_snapshot_download_stalled", "TimeoutError"); + const idle = new AbortController(); + const signal = input.signal ? AbortSignal.any([input.signal, idle.signal]) : idle.signal; + const timer = setTimeout(() => idle.abort(stalled), DOWNLOAD_IDLE_TIMEOUT_MS); + timer.unref(); + try { + const object = await input.storage.getObject(ref.companyId, ref.objectKey, { signal }); + const bounded = new Transform({ transform(bytes, _encoding, callback) { + if (bytes.length > 0) timer.refresh(); + received += bytes.length; digest.update(bytes); + callback(received > ref.byteSize ? new Error("remote_workspace_snapshot_object_size_mismatch") : null, bytes); + } }); + try { await pipeline(object.stream, bounded, createWriteStream(ciphertext, { flags: "wx", mode: 0o600 }), { signal }); } + finally { object.stream.destroy(); } + } catch (error) { + if (signal.aborted && signal.reason === stalled) throw stalled; + throw error; + } finally { clearTimeout(timer); } if (received !== ref.byteSize || digest.digest("hex") !== ref.sha256) failure("object_hash_mismatch"); const file = await open(ciphertext, "r"), header = Buffer.alloc(MAGIC.length + 12), tag = Buffer.alloc(16); try { await file.read(header, 0, header.length, 0); await file.read(tag, 0, tag.length, ref.byteSize - tag.length); } finally { await file.close(); } diff --git a/server/src/services/remote-workspace-transport.ts b/server/src/services/remote-workspace-transport.ts index f46734cca37..04ef461726a 100644 --- a/server/src/services/remote-workspace-transport.ts +++ b/server/src/services/remote-workspace-transport.ts @@ -208,7 +208,7 @@ const WORKSPACE_COMMAND_TIMEOUT_MS = 300_000; async function command(target: RemoteWorkspaceTarget, operation: RemoteWorkspaceOperation, args: string[], env?: Record, timeoutMs = WORKSPACE_COMMAND_TIMEOUT_MS) { const execute = () => target.runner!.execute({ command: "node", args, cwd: target.remoteCwd, - env: { NODE_OPTIONS: "", NODE_PATH: "", ...env }, timeoutMs, bypassSession: true }); + env: { NODE_OPTIONS: "", NODE_PATH: "", ...env, NODE_NO_WARNINGS: "1" }, timeoutMs, bypassSession: true }); // Copy-back and restore record why the provider refused the command. const result = operation === "capture" || operation === "archive_helper" || operation === "restore" ? await execute().catch((error: unknown) => { noteProviderRefusal(operation, error); throw error; }) diff --git a/server/src/services/remote-workspaces.integration.test.ts b/server/src/services/remote-workspaces.integration.test.ts index 0ce20d415c9..1037afd1bbf 100644 --- a/server/src/services/remote-workspaces.integration.test.ts +++ b/server/src/services/remote-workspaces.integration.test.ts @@ -5,6 +5,7 @@ import { createHash, randomUUID } from "node:crypto"; import { execFile, spawn } from "node:child_process"; import net from "node:net"; import { createServer } from "node:http"; +import { Readable } from "node:stream"; import { promisify } from "node:util"; import fs from "node:fs/promises"; import os from "node:os"; @@ -406,7 +407,7 @@ suite("remote workspace lifecycle with real Postgres and independent filesystem } }, 60_000); - it.each(["same root", "different root", "destroyed allocation", "occupied root", "missing object", "transient restore", "restore lock busy"] as const)( + it.each(["same root", "different root", "destroyed allocation", "occupied root", "missing object", "stalled download", "transient restore", "restore lock busy"] as const)( "preserves the published workspace path and files on a cold restore (%s)", async mode => { const reserved = await reserve({ kind: "empty" }), first = await target("sandbox-A"); const prepared = await prepareRemoteWorkspace({ db, storage, workspace: reserved, runId: ids.run, lease: await lease(), target: first }); @@ -512,6 +513,29 @@ suite("remote workspace lifecycle with real Postgres and independent filesystem message: "remote_workspace_restore_failed: step=snapshot_download; remote_workspace_snapshot_object_missing" }); expect(await fs.lstat(expectedRemoteRoot).catch(() => null)).toBeNull(); } + if (mode === "stalled download") { + let ready!: () => void; + const started = new Promise(resolve => { ready = resolve; }); + const body = new Readable({ read() {} }); + const stalledStorage: StorageService = { ...storage, getObject: async () => { ready(); return { stream: body }; } }; + const realTimeout = globalThis.setTimeout; + const deadline = vi.spyOn(globalThis, "setTimeout").mockImplementation((fn, ms, ...args) => + realTimeout(fn, ms === 300_000 ? 25 : ms, ...args)); + const restoring = prepareRemoteWorkspace({ db, storage: stalledStorage, workspace: reservedAgain, runId: nextRun, lease: newLease, target: second }) + .then(() => null, (error: unknown) => error as RemoteWorkspaceRestoreError); + try { + await started; + expect(await restoring).toMatchObject({ step: "snapshot_download", retryable: true, + message: "remote_workspace_restore_failed: step=snapshot_download; remote_workspace_snapshot_download_stalled" }); + expect(body.destroyed).toBe(true); + expect(readRemoteWorkspaceAuthority((await workspace(reserved.id)).metadata)) + .toEqual(readRemoteWorkspaceAuthority(reservedAgain.metadata)); + expect(await fs.lstat(expectedRemoteRoot).catch(() => null)).toBeNull(); + } finally { + body.destroy(new Error("test cleanup")); await restoring; + deadline.mockRestore(); + } + } const restored = await prepareRemoteWorkspace({ db, storage, workspace: reservedAgain, runId: nextRun, lease: newLease, target: second }); expect(restored.remoteRoot).toBe(expectedRemoteRoot); if (mode !== "different root") expect(restored.remoteRoot).toBe(prepared.remoteRoot); @@ -974,10 +998,12 @@ suite("remote workspace lifecycle with real Postgres and independent filesystem for (const mode of ["abort", "storage_failure"] as const) { const controller = new AbortController(), failure = new Error("Saved object unavailable"); let observedSignal: AbortSignal | undefined; + let observedBody: Readable | undefined; const interruptedStorage: StorageService = { ...storage, getObject: async (companyId, objectKey, options) => { observedSignal = options?.signal; if (mode === "storage_failure") throw failure; const object = await storage.getObject(companyId, objectKey, options); + observedBody = object.stream; controller.abort(new DOMException("Saved read cancelled", "AbortError")); return object; } }; @@ -986,9 +1012,15 @@ suite("remote workspace lifecycle with real Postgres and independent filesystem const access = await remoteFiles.open({ companyId: ids.company, workspaceId: reserved.id, signal: controller.signal }) .catch(error => { readError = error; return null; }); await access?.close(); - expect(observedSignal).toBe(controller.signal); + expect(observedSignal).toBeInstanceOf(AbortSignal); + expect(observedSignal?.aborted).toBe(controller.signal.aborted); if (mode === "storage_failure") expect(readError).toBe(failure); - else expect(readError).toMatchObject({ name: "AbortError" }); + else { + expect(observedSignal?.reason).toBe(controller.signal.reason); + expect(readError).toMatchObject({ name: "AbortError" }); + expect(observedBody?.destroyed).toBe(true); + expect(observedBody?.closed).toBe(true); + } expect(await fs.readdir(savedTemp)).toEqual([]); } const cancelled = new DOMException("Already cancelled", "AbortError");