diff --git a/server/src/services/remote-workspace-transport.ts b/server/src/services/remote-workspace-transport.ts index 3e821827f68..59df98adf52 100644 --- a/server/src/services/remote-workspace-transport.ts +++ b/server/src/services/remote-workspace-transport.ts @@ -398,9 +398,29 @@ export async function cleanupRemoteWorkspaceArchive(input: { await command(input.target, "cleanup", ["-e", script, input.remoteCwd, input.operationId], undefined, 10_000); } +/** + * Wall-clock milliseconds for each save phase, filled in as each phase ends so a + * failed save still reports the phases it finished. It holds only durations and + * archive counts, never paths or provider text. + */ +export type RemoteWorkspaceSavePhaseTiming = { + /** Archive helper install plus in-sandbox capture. */ + captureMs?: number; + /** Provider download of the captured archive. */ + transferMs?: number; + /** Server-side re-hash of the downloaded archive. */ + verifyMs?: number; + byteSize?: number; + entries?: number; +}; + +const elapsedSince = (started: number) => Math.max(0, Math.round(performance.now() - started)); + export async function captureRemoteWorkspaceArchive(input: { - target: RemoteWorkspaceTarget; remoteCwd: string; operationId: string; + target: RemoteWorkspaceTarget; remoteCwd: string; operationId: string; timing?: RemoteWorkspaceSavePhaseTiming; }) { + const timing = input.timing ?? {}; + const captureStarted = performance.now(); const staged = await installArchiveHelper(input.target, input.remoteCwd, input.operationId); // Publish the archive and receipt as one immutable directory. Concurrent // invocations own separate attempts; no invocation unlinks another's archive. @@ -444,14 +464,20 @@ export async function captureRemoteWorkspaceArchive(input: { }; if (!["legacy", "captured"].includes(layout) || !/^[a-f0-9]{64}$/.test(receipt.manifestDigest) || !/^[a-f0-9]{64}$/.test(receipt.sha256) || !Number.isSafeInteger(receipt.byteSize) || receipt.byteSize <= 0) throw new Error("remote_workspace_capture_receipt_invalid"); + timing.captureMs = elapsedSince(captureStarted); timing.byteSize = receipt.byteSize; + if (Number.isSafeInteger(receipt.entries) && receipt.entries >= 0) timing.entries = receipt.entries; const root = await mkdtemp(path.join(os.tmpdir(), "paperclip-workspace-upload-")); const archivePath = path.join(root, "workspace.pcws"); try { + const transferStarted = performance.now(); await transferArchive(input.target, "syncOut", [{ operationId: input.operationId, files: [{ kind: "file", sourcePath: layout === "captured" ? path.posix.join(staged.directory, "captured", "workspace.pcws") : staged.archive, targetPath: archivePath }], }], receipt.byteSize); + timing.transferMs = elapsedSince(transferStarted); + const verifyStarted = performance.now(); const hash = createHash("sha256"); let bytes = 0; for await (const chunk of createReadStream(archivePath)) { hash.update(chunk); bytes += chunk.length; } + timing.verifyMs = elapsedSince(verifyStarted); if (hash.digest("hex") !== receipt.sha256 || bytes !== receipt.byteSize) throw new Error("remote_workspace_capture_transfer_mismatch"); return { ...receipt, archivePath, cleanup: () => rm(root, { recursive: true, force: true }) }; } catch (error) { diff --git a/server/src/services/remote-workspaces.integration.test.ts b/server/src/services/remote-workspaces.integration.test.ts index 7b8e13bf206..8c46f3edb43 100644 --- a/server/src/services/remote-workspaces.integration.test.ts +++ b/server/src/services/remote-workspaces.integration.test.ts @@ -1115,9 +1115,26 @@ suite("remote workspace lifecycle with real Postgres and independent filesystem expect(readRemoteWorkspaceAuthority((await workspace(reserved.id)).metadata)).toMatchObject({ dirty: true, writerRunId: ids.run, snapshotId: readRemoteWorkspaceAuthority(prepared.workspace.metadata)!.snapshotId }); expect((await fs.readdir(path.join(prepared.remoteRoot, ".paperclip-runtime", "workspace-snapshots"))).length).toBe(1); - const saved = await saveRemoteWorkspace({ db, storage, workspace: await workspace(reserved.id), runId: ids.run, + // Each settled save records where its time went: the failed upload keeps the + // phases it finished, and the retry of the same boundary records every phase. + const finalTiming = async () => (await db.select().from(workspaceOperations).where(and( + eq(workspaceOperations.executionWorkspaceId, reserved.id), eq(workspaceOperations.phase, "remote_workspace_snapshot"), + sql`${workspaceOperations.metadata}->>'boundary' = 'final'`))).map(save => save.metadata?.saveTiming); + const duration = expect.any(Number), archive = { byteSize: expect.any(Number), entries: expect.any(Number) }; + expect(await finalTiming()).toEqual([{ schema: "paperclip.remote-workspace-save-timing.v1", outcome: "failed", totalMs: duration, + captureMs: duration, transferMs: duration, verifyMs: duration, ...archive, at: expect.any(String) }]); + const beforeSuccess = await workspace(reserved.id); + const saved = await saveRemoteWorkspace({ db, storage, workspace: beforeSuccess, runId: ids.run, lease: prepared.lease, target: first, assertOwned: async () => undefined }); expect(saved.generation).toBe(2); + const [succeededTiming] = await finalTiming(); + // A lost-reply replay of the committed save keeps the measured record. + await expect(saveRemoteWorkspace({ db, storage, workspace: beforeSuccess, runId: ids.run, + lease: prepared.lease, target: first, assertOwned: async () => undefined })).resolves.toMatchObject({ snapshotId: saved.snapshotId }); + expect(await finalTiming()).toEqual([succeededTiming]); + expect(await finalTiming()).toEqual([{ schema: "paperclip.remote-workspace-save-timing.v1", outcome: "succeeded", totalMs: duration, + captureMs: duration, transferMs: duration, verifyMs: duration, uploadMs: duration, publishMs: duration, ...archive, + at: expect.any(String) }]); expect(await fs.readdir(path.join(prepared.remoteRoot, ".paperclip-runtime", "workspace-snapshots"))).toEqual([]); }, 60_000); @@ -1224,9 +1241,12 @@ suite("remote workspace lifecycle with real Postgres and independent filesystem .rejects.toThrow("remote_workspace_snapshot_operation_binding_changed"); expect(upload).not.toHaveBeenCalled(); const [held] = await db.select().from(workspaceOperations).where(eq(workspaceOperations.id, poisoned!.id)); - const { saveWork, ...originalMetadata } = held!.metadata!; + const { saveWork, saveTiming, ...originalMetadata } = held!.metadata!; expect(originalMetadata).toEqual(poisoned!.metadata); expect(saveWork).toMatchObject({ activeIds: [], unknown: true, quiescedAt: expect.any(String) }); + // The refused invocation records only its own failed duration; it never captured or uploaded. + expect(saveTiming).toEqual({ schema: "paperclip.remote-workspace-save-timing.v1", outcome: "failed", + totalMs: expect.any(Number), at: expect.any(String) }); expect(Number.isFinite(Date.parse((saveWork as { quiescedAt: string }).quiescedAt))).toBe(true); expect(await db.select().from(executionWorkspaceSnapshots).where(eq(executionWorkspaceSnapshots.sourceOperationId, poisoned!.id))) .toHaveLength(0); diff --git a/server/src/services/remote-workspaces.ts b/server/src/services/remote-workspaces.ts index 8af46c45c4e..63aa0bfe1e1 100644 --- a/server/src/services/remote-workspaces.ts +++ b/server/src/services/remote-workspaces.ts @@ -28,6 +28,7 @@ import { captureRemoteWorkspaceArchive, cleanupRemoteWorkspaceArchive, prepareRemoteWorkspaceDirectory, prepareRemoteRepository, remoteWorkspaceRepositoryPath, RemoteWorkspaceOperationTimeoutError, RemoteWorkspacePreparationTimeoutError, remoteWorkspaceFailureCause, requireRemoteWorkspaceTarget, restoreRemoteWorkspaceDirectory, RemoteWorkspaceRestoreError, type RemoteWorkspaceRestoreStep, + type RemoteWorkspaceSavePhaseTiming, } from "./remote-workspace-transport.js"; import { beginWorkspaceSnapshot, getWorkspaceSnapshot, loadWorkspaceSnapshotArchive, uploadWorkspaceSnapshot, @@ -46,6 +47,16 @@ type Workspace = Pick[0]>[0]; const object = (value: unknown): Record => value && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; const SAVE_WORK_SCHEMA = "paperclip.remote-workspace-save-work.v1"; +const SAVE_TIMING_SCHEMA = "paperclip.remote-workspace-save-timing.v1"; + +/** Per-phase durations of the latest settled save invocation, kept in the operation's local metadata. */ +type RemoteWorkspaceSaveTiming = RemoteWorkspaceSavePhaseTiming & { + /** Encrypt, hash, and store the archive (zero work when a prior upload is reused). */ + uploadMs?: number; + /** Read the archive manifest and commit the saved authority. */ + publishMs?: number; + reusedUpload?: boolean; +}; const RETIRED_SAVE_SCHEMA = "paperclip.remote-workspace-retired-save.v1"; type SaveWork = { schema: typeof SAVE_WORK_SCHEMA; activeIds: string[]; unknown: boolean; quiescedAt: string | null }; function readSaveWork(metadata: Record | null | undefined): SaveWork | null { @@ -610,9 +621,15 @@ export function remoteWorkspaceSaveFailure(error: unknown): { code: string; caus return cause ? { code, cause } : { code }; } +function saveTimingRecord(timing: RemoteWorkspaceSaveTiming & { startedAt: number }, outcome: "succeeded" | "failed", at: Date) { + const { startedAt, ...phases } = timing; + return { schema: SAVE_TIMING_SCHEMA, outcome, totalMs: Math.max(0, Math.round(performance.now() - startedAt)), + ...phases, at: at.toISOString() }; +} + async function settleSaveInvocation(input: { db: Db; workspace: Workspace; runId: string; epoch: number; location: NonNullable; operationId: string; invocationId: string; - failure?: { code: string; cause?: string } }) { + failure?: { code: string; cause?: string }; timing?: RemoteWorkspaceSaveTiming & { startedAt: number } }) { return input.db.transaction(async tx => { const workspace = await currentWorkspace(tx, input.workspace.companyId, input.workspace.id); const [operation] = await tx.select().from(workspaceOperations).where(and(eq(workspaceOperations.id, input.operationId), @@ -629,6 +646,7 @@ async function settleSaveInvocation(input: { db: Db; workspace: Workspace; runId const failed = Boolean(failure) && operation.status === "running" && !activeIds.length && !work.unknown; await tx.update(workspaceOperations).set({ metadata: { ...operation.metadata, saveWork: { ...work, activeIds, quiescedAt: activeIds.length ? null : now.toISOString() }, + ...(input.timing ? { saveTiming: saveTimingRecord(input.timing, input.failure ? "failed" : "succeeded", now) } : {}), ...(failed && failure ? { lastFailure: { ...failure, at: now.toISOString() } } : {}) }, ...(failed && failure ? { status: "failed", exitCode: 1, finishedAt: now, stderrExcerpt: `${[failure.code, failure.cause].filter(Boolean).join("\n")}\n` } : {}), @@ -644,6 +662,7 @@ export async function saveRemoteWorkspace(input: { assertOwned: () => Promise; storage?: StorageService; boundary?: string; repositories?: NonNullable; }) { + const saveStarted = performance.now(); const target = requireRemoteWorkspaceTarget(input.target), boundary = input.boundary ?? "final"; const expected = readRemoteWorkspaceAuthority(input.workspace.metadata); if (!expected?.location || expected.location.providerLeaseId !== input.lease.providerLeaseId || expected.location.leaseId !== input.lease.id) throw new Error("remote_workspace_save_binding_mismatch"); @@ -696,21 +715,31 @@ export async function saveRemoteWorkspace(input: { let captured: Awaited> | null = null; let publicationScratch: string | undefined; let failure: { code: string; cause?: string } | undefined; + const timing: RemoteWorkspaceSaveTiming & { startedAt: number } = { startedAt: saveStarted }; + // A replay of an already-committed save measures no phase; it keeps the + // original save's timing instead of overwriting it with an empty record. + let replayedCommittedSave = false; try { const snapshot = await beginWorkspaceSnapshot(input.db, { companyId: input.workspace.companyId, workspaceId: input.workspace.id, operationId: operation.id, runId: input.runId, parentSnapshotId: operation.metadata?.parentSnapshotId as string | null, generation: operation.metadata?.generation as number, epoch: expected.epoch }); if (operation.status === "succeeded" && ["committed", "superseded"].includes(snapshot.status)) { + replayedCommittedSave = true; return { snapshotId: snapshot.id, generation: snapshot.generation, sha256: snapshot.sha256!, savedAt: snapshot.updatedAt.toISOString() }; } const existingUpload = snapshot.status === "uploaded" ? await getWorkspaceSnapshot(input.db, { companyId: input.workspace.companyId, workspaceId: input.workspace.id, snapshotId: snapshot.id }) : null; - captured = existingUpload ? null : await captureRemoteWorkspaceArchive({ target, remoteCwd: expected.location.remoteCwd, operationId: operation.id }); + if (existingUpload) timing.reusedUpload = true; + captured = existingUpload ? null : await captureRemoteWorkspaceArchive({ target, remoteCwd: expected.location.remoteCwd, + operationId: operation.id, timing }); await input.assertOwned(); + const uploadStarted = performance.now(); const saved = await uploadWorkspaceSnapshot({ db: input.db, storage: input.storage ?? getStorageService(), snapshotId: snapshot.id, companyId: input.workspace.companyId, workspaceId: input.workspace.id, archivePath: captured?.archivePath ?? "", manifestDigest: existingUpload?.manifestDigest ?? captured!.manifestDigest }); + timing.uploadMs = Math.max(0, Math.round(performance.now() - uploadStarted)); await input.assertOwned(); + const publishStarted = performance.now(); let publicationArchive = captured?.archivePath; if (!publicationArchive) { publicationScratch = await mkdtemp(path.join(os.tmpdir(), "pc-workspace-publication-")); @@ -756,6 +785,7 @@ export async function saveRemoteWorkspace(input: { deleteAfter: new Date(at.getTime() + 7 * 86400_000), updatedAt: at }).where(and(eq(executionWorkspaceSnapshots.id, snapshot.parentSnapshotId), eq(executionWorkspaceSnapshots.companyId, workspace.companyId), eq(executionWorkspaceSnapshots.executionWorkspaceId, workspace.id))); }); + timing.publishMs = Math.max(0, Math.round(performance.now() - publishStarted)); return { snapshotId: saved.id, generation: saved.generation, sha256: saved.sha256, savedAt: at.toISOString() }; } catch (error) { failure = remoteWorkspaceSaveFailure(error); @@ -768,7 +798,8 @@ export async function saveRemoteWorkspace(input: { catch { logger.warn({ runId: input.runId, operationId: operation.id }, "Remote workspace local archive cleanup failed"); } try { if (await settleSaveInvocation({ db: input.db, workspace: input.workspace, runId: input.runId, epoch: expected.epoch, - location: expected.location, operationId: operation.id, invocationId, failure })) { + location: expected.location, operationId: operation.id, invocationId, failure, + timing: replayedCommittedSave ? undefined : timing })) { await cleanupSavedExports({ db: input.db, companyId: input.workspace.companyId, workspaceId: input.workspace.id, location: expected.location, target, assertOwned: input.assertOwned }); }