diff --git a/packages/adapter-utils/src/runtime-progress.test.ts b/packages/adapter-utils/src/runtime-progress.test.ts index ee0800b71db..24048404e73 100644 --- a/packages/adapter-utils/src/runtime-progress.test.ts +++ b/packages/adapter-utils/src/runtime-progress.test.ts @@ -205,4 +205,77 @@ describe("createRuntimeProgressReporter", () => { "[paperclip] Restoring workspace from ssh: 8.0 MB\n", ]); }); + + it("fail() emits a terminal failure marker with the last percent instead of a dangling line", async () => { + const clock = makeClock(); + const lines: string[] = []; + const reporter = createRuntimeProgressReporter({ + sink: (line) => { + lines.push(line); + }, + phase: "Syncing", + label: "workspace", + direction: "to", + target: "ssh", + now: clock.now, + }); + + await reporter.report(40 * MB, 100 * MB); // emit at 40% + await reporter.fail(); + + expect(lines).toEqual([ + "[paperclip] Syncing workspace to ssh: 40% (40.0/100.0 MB)\n", + "[paperclip] Syncing workspace to ssh: failed at 40% (40.0/100.0 MB)\n", + ]); + }); + + it("fail() falls back to an MB marker when the total is unknown", async () => { + const lines: string[] = []; + const reporter = createRuntimeProgressReporter({ + sink: (line) => { + lines.push(line); + }, + phase: "Restoring", + direction: "from", + target: "ssh", + }); + + await reporter.report(3 * MB, null); + await reporter.fail(); + + expect(lines.at(-1)).toBe("[paperclip] Restoring from ssh: failed after 3.0 MB\n"); + }); + + it("fail() is suppressed after a terminal completion and complete() after a failure", async () => { + const lines: string[] = []; + const reporter = createRuntimeProgressReporter({ + sink: (line) => { + lines.push(line); + }, + phase: "Syncing", + label: "workspace", + direction: "to", + target: "ssh", + }); + + await reporter.report(100 * MB, 100 * MB); // terminal complete + await reporter.fail(); // suppressed — already completed + expect(lines).toHaveLength(1); + + const failLines: string[] = []; + const failed = createRuntimeProgressReporter({ + sink: (line) => { + failLines.push(line); + }, + phase: "Syncing", + label: "workspace", + direction: "to", + target: "ssh", + }); + await failed.report(20 * MB, 100 * MB); + await failed.fail(); + await failed.complete(); // suppressed — already failed + expect(failLines).toHaveLength(2); + expect(failLines.at(-1)).toContain("failed at 20%"); + }); }); diff --git a/packages/adapter-utils/src/runtime-progress.ts b/packages/adapter-utils/src/runtime-progress.ts index 09f12d982a6..8cdd7ce4ec2 100644 --- a/packages/adapter-utils/src/runtime-progress.ts +++ b/packages/adapter-utils/src/runtime-progress.ts @@ -47,6 +47,12 @@ export interface RuntimeProgressReporter { * Emit the terminal completion line if it hasn't been emitted yet. Idempotent. */ complete(doneBytes?: number, totalBytes?: number | null): Promise; + /** + * Emit a terminal failure line if no terminal line has been emitted yet, so a + * failed transfer leaves an explicit marker instead of a dangling percentage. + * Idempotent and mutually exclusive with `complete()`. + */ + fail(doneBytes?: number, totalBytes?: number | null): Promise; } const BYTES_PER_MB = 1024 * 1024; @@ -83,6 +89,14 @@ export function createRuntimeProgressReporter( return `${prefix}: ${formatMb(doneBytes)} MB\n`; } + function buildFailLine(doneBytes: number, totalBytes: number | null): string { + if (totalBytes != null && totalBytes > 0) { + const pct = clampPercent((doneBytes / totalBytes) * 100); + return `${prefix}: failed at ${pct}% (${formatMb(doneBytes)}/${formatMb(totalBytes)} MB)\n`; + } + return `${prefix}: failed after ${formatMb(doneBytes)} MB\n`; + } + async function emit(doneBytes: number, totalBytes: number | null): Promise { lastEmitAt = now(); if (totalBytes != null && totalBytes > 0) { @@ -127,5 +141,12 @@ export function createRuntimeProgressReporter( : lastDoneBytes; await options.sink(buildLine(done, total)); }, + async fail(doneBytes, totalBytes) { + if (completed) return; + completed = true; + const total = totalBytes !== undefined ? totalBytes : lastTotalBytes; + const done = doneBytes !== undefined ? doneBytes : lastDoneBytes; + await options.sink(buildFailLine(done, total)); + }, }; } diff --git a/packages/adapter-utils/src/ssh.ts b/packages/adapter-utils/src/ssh.ts index ab8df93a4f1..d3efffd7252 100644 --- a/packages/adapter-utils/src/ssh.ts +++ b/packages/adapter-utils/src/ssh.ts @@ -488,6 +488,8 @@ interface TransferProgress { transferred: () => number; // Emit the terminal completion line. Idempotent. finish: () => Promise; + // Emit a terminal failure marker instead of a completion line. Idempotent. + fail: () => Promise; } // Wraps a throttled progress reporter behind a counting Transform so transports @@ -496,12 +498,18 @@ interface TransferProgress { // is an estimate (tar upload / remote probe) we clamp the reported bytes to 99% // of the estimate so an inaccurate total never shows a premature 100%; `finish` // then emits the terminal 100% (or, in MB-only mode, the final MB) line. +// +// `totalBytes` may be a promise so an expensive size estimate (a local dir walk +// or a remote `du` probe) runs concurrently with the transfer instead of +// blocking the pipe from opening. Until it resolves the counter reports bytes in +// MB-only mode, then adopts the percentage once the total is known; `finish` +// awaits the estimate so the terminal 100% line is still guaranteed. function createTransferProgress(input: { onProgress: RuntimeProgressSink; phase: RuntimeProgressPhase; direction: RuntimeProgressDirection; label?: string; - totalBytes: number | null; + totalBytes: number | null | Promise; estimated: boolean; }): TransferProgress { const reporter = createRuntimeProgressReporter({ @@ -511,8 +519,17 @@ function createTransferProgress(input: { label: input.label, target: "ssh", }); - const total = input.totalBytes != null && input.totalBytes > 0 ? input.totalBytes : null; - const cap = total != null && input.estimated ? Math.floor(total * 0.99) : null; + + let total: number | null = null; + let cap: number | null = null; + const applyTotal = (value: number | null) => { + total = value != null && value > 0 ? value : null; + cap = total != null && input.estimated ? Math.floor(total * 0.99) : null; + }; + const totalReady: Promise = + input.totalBytes != null && typeof (input.totalBytes as Promise).then === "function" + ? (input.totalBytes as Promise).then(applyTotal, () => applyTotal(null)) + : (applyTotal(input.totalBytes as number | null), Promise.resolve()); let transferred = 0; let chain: Promise = Promise.resolve(); @@ -524,7 +541,8 @@ function createTransferProgress(input: { transform(chunk: Buffer, _encoding, callback) { transferred += chunk.length; const reported = cap != null ? Math.min(transferred, cap) : transferred; - enqueue(() => reporter.report(reported, total)); + const totalSnapshot = total; + enqueue(() => reporter.report(reported, totalSnapshot)); callback(null, chunk); }, }); @@ -534,8 +552,13 @@ function createTransferProgress(input: { transferred: () => transferred, finish: async () => { await chain.catch(() => undefined); + await totalReady.catch(() => undefined); await reporter.complete(total != null ? total : transferred, total).catch(() => undefined); }, + fail: async () => { + await chain.catch(() => undefined); + await reporter.fail(transferred, total).catch(() => undefined); + }, }; } @@ -784,13 +807,18 @@ async function importGitWorkspaceToSsh(input: { }) : null; - await streamLocalFileToSsh({ - spec: input.spec, - localFile: bundlePath, - remoteScript: remoteSetupScript, - progress: progress ?? undefined, - }); - await progress?.finish(); + try { + await streamLocalFileToSsh({ + spec: input.spec, + localFile: bundlePath, + remoteScript: remoteSetupScript, + progress: progress ?? undefined, + }); + await progress?.finish(); + } catch (error) { + await progress?.fail(); + throw error; + } } finally { await runLocalGit(input.localDir, ["update-ref", "-d", tempRef], { timeout: 10_000, @@ -836,13 +864,18 @@ async function exportGitWorkspaceFromSsh(input: { }) : null; - await streamSshToLocalFile({ - spec: input.spec, - remoteScript: exportScript, - localFile: bundlePath, - progress: progress ?? undefined, - }); - await progress?.finish(); + try { + await streamSshToLocalFile({ + spec: input.spec, + remoteScript: exportScript, + localFile: bundlePath, + progress: progress ?? undefined, + }); + await progress?.finish(); + } catch (error) { + await progress?.fail(); + throw error; + } await runLocalGit(input.localDir, ["fetch", "--force", bundlePath, `refs/paperclip/ssh-sync/export:${importedRef}`], { timeout: 60_000, @@ -1236,13 +1269,15 @@ export async function syncDirectoryToSsh(input: { // tar's archive size isn't known until tar finishes, so estimate it from the // local file sizes and clamp the reported percent to 99% until the pipe closes. + // The estimate walk runs concurrently with the transfer so it never delays the + // pipe from opening on large workspaces. const progress = input.onProgress ? createTransferProgress({ onProgress: input.onProgress, phase: "Syncing", direction: "to", label: input.progressLabel, - totalBytes: await estimateLocalDirSize({ + totalBytes: estimateLocalDirSize({ localDir: input.localDir, exclude: input.exclude, followSymlinks: input.followSymlinks, @@ -1251,7 +1286,8 @@ export async function syncDirectoryToSsh(input: { }) : null; - await new Promise((resolve, reject) => { + try { + await new Promise((resolve, reject) => { const tarArgs = [ ...(input.followSymlinks ? ["-h"] : []), "-C", @@ -1328,8 +1364,12 @@ export async function syncDirectoryToSsh(input: { sshExitCode = code; maybeFinish(); }); - }).finally(auth.cleanup); - await progress?.finish(); + }).finally(auth.cleanup); + await progress?.finish(); + } catch (error) { + await progress?.fail(); + throw error; + } } export async function syncDirectoryFromSsh(input: { @@ -1356,19 +1396,17 @@ export async function syncDirectoryFromSsh(input: { ]; // The remote tar size isn't known locally, so probe the remote directory for - // an estimate (clamped to 99%). When the probe is unavailable we report bytes - // received in MB mode with a terminal completion line. - const remoteTotal = input.onProgress - ? await probeRemoteDirSize({ spec: input.spec, remoteDir: input.remoteDir }) - : null; + // an estimate (clamped to 99%). The probe runs concurrently with the transfer + // so its round-trip never delays the restore; when it is unavailable we report + // bytes received in MB mode with a terminal completion line. const progress = input.onProgress ? createTransferProgress({ onProgress: input.onProgress, phase: "Restoring", direction: "from", label: input.progressLabel, - totalBytes: remoteTotal, - estimated: remoteTotal != null, + totalBytes: probeRemoteDirSize({ spec: input.spec, remoteDir: input.remoteDir }), + estimated: true, }) : null; @@ -1442,6 +1480,9 @@ export async function syncDirectoryFromSsh(input: { await clearLocalDirectory(input.localDir, input.preserveLocalEntries); await copyDirectoryContents(stagingDir, input.localDir); + } catch (error) { + await progress?.fail(); + throw error; } finally { await fs.rm(stagingDir, { recursive: true, force: true }).catch(() => undefined); await auth.cleanup();