mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-25 09:41:04 +01:00
fix(adapter-utils): ensure sync timers always finish and sink errors do not propagate
Use try/finally around each sync stage group so timer.finish() is called even when tar, upload, or extract throws. Wrap the sink call in finish() with try/catch so a rejecting progress sink cannot fail a completed sync. Add test covering the sink-rejection case (31/31 pass). Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
3 files changed
+153
-124
No files matched your search
@@ -385,4 +385,20 @@ describe("createSyncStageTimer", () => {
|
||||
await disabled.time("tar", async () => {});
|
||||
await expect(disabled.finish()).resolves.toBeUndefined();
|
||||
});
|
||||
|
||||
it("swallows a rejecting sink so a logging failure cannot fail a completed sync", async () => {
|
||||
const clock = makeClock();
|
||||
const timer = createSyncStageTimer({
|
||||
sink: async () => {
|
||||
throw new Error("sink rejected");
|
||||
},
|
||||
phase: "git_sync",
|
||||
artifact: "git history",
|
||||
now: clock.now,
|
||||
});
|
||||
clock.advance(5);
|
||||
await timer.time("tar", async () => {});
|
||||
// finish() must resolve even though the sink rejects.
|
||||
await expect(timer.finish()).resolves.toBeUndefined();
|
||||
});
|
||||
});
|
||||
@@ -101,9 +101,13 @@ export function createSyncStageTimer(options: SyncStageTimerOptions): SyncStageT
|
||||
if (!options.sink || stages.length === 0) return;
|
||||
const totalMs = stages.reduce((sum, entry) => sum + entry.durationMs, 0);
|
||||
const breakdown = stages.map((entry) => `${entry.stage} ${entry.durationMs}ms`).join(", ");
|
||||
await options.sink(
|
||||
`[paperclip] sync ${options.phase} ${options.artifact}: ${breakdown} (total ${totalMs}ms) [transport=${transport}]\n`,
|
||||
);
|
||||
try {
|
||||
await options.sink(
|
||||
`[paperclip] sync ${options.phase} ${options.artifact}: ${breakdown} (total ${totalMs}ms) [transport=${transport}]\n`,
|
||||
);
|
||||
} catch {
|
||||
// Sink errors must not propagate — duration logging is observability-only.
|
||||
}
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
@@ -539,43 +539,46 @@ export async function prepareSandboxManagedRuntime(input: {
|
||||
artifact: "git history",
|
||||
transport: syncTransport,
|
||||
});
|
||||
await withShallowGitWorkspaceClone({
|
||||
localDir: input.workspaceLocalDir,
|
||||
snapshot: gitSnapshot,
|
||||
}, async (cloneDir) => {
|
||||
const gitTarPath = path.join(tempDir, "git-workspace.tar");
|
||||
const gitTarBytes = await gitTimer.time("tar", async () => {
|
||||
await createTarballFromDirectory({
|
||||
localDir: cloneDir,
|
||||
archivePath: gitTarPath,
|
||||
exclude: [".paperclip-runtime"],
|
||||
try {
|
||||
await withShallowGitWorkspaceClone({
|
||||
localDir: input.workspaceLocalDir,
|
||||
snapshot: gitSnapshot,
|
||||
}, async (cloneDir) => {
|
||||
const gitTarPath = path.join(tempDir, "git-workspace.tar");
|
||||
const gitTarBytes = await gitTimer.time("tar", async () => {
|
||||
await createTarballFromDirectory({
|
||||
localDir: cloneDir,
|
||||
archivePath: gitTarPath,
|
||||
exclude: [".paperclip-runtime"],
|
||||
});
|
||||
return fs.readFile(gitTarPath);
|
||||
});
|
||||
return fs.readFile(gitTarPath);
|
||||
const remoteGitTar = path.posix.join(runtimeRootDir, "git-workspace-upload.tar");
|
||||
await input.client.makeDir(runtimeRootDir);
|
||||
const gitUpload = makeTransferProgress(
|
||||
input.onProgress,
|
||||
"Syncing",
|
||||
"to",
|
||||
"git history",
|
||||
{ sink: input.onRuntimeProgress, phase: "git_sync" },
|
||||
);
|
||||
await gitTimer.time("upload", async () => {
|
||||
await input.client.writeFile(remoteGitTar, toArrayBuffer(gitTarBytes), gitUpload.options);
|
||||
await gitUpload.finish(gitTarBytes.byteLength, gitTarBytes.byteLength);
|
||||
});
|
||||
await gitTimer.time("extract", () => input.client.run(
|
||||
`sh -c ${shellQuote(
|
||||
`mkdir -p ${shellQuote(workspaceRemoteDir)} && ` +
|
||||
`find ${shellQuote(workspaceRemoteDir)} -mindepth 1 -maxdepth 1 ${preserveFindArgs([".paperclip-runtime"])} -exec rm -rf -- {} + && ` +
|
||||
`tar -xf ${shellQuote(remoteGitTar)} -C ${shellQuote(workspaceRemoteDir)} && ` +
|
||||
`rm -f ${shellQuote(remoteGitTar)}`,
|
||||
)}`,
|
||||
{ timeoutMs: input.spec.timeoutMs },
|
||||
));
|
||||
});
|
||||
const remoteGitTar = path.posix.join(runtimeRootDir, "git-workspace-upload.tar");
|
||||
await input.client.makeDir(runtimeRootDir);
|
||||
const gitUpload = makeTransferProgress(
|
||||
input.onProgress,
|
||||
"Syncing",
|
||||
"to",
|
||||
"git history",
|
||||
{ sink: input.onRuntimeProgress, phase: "git_sync" },
|
||||
);
|
||||
await gitTimer.time("upload", async () => {
|
||||
await input.client.writeFile(remoteGitTar, toArrayBuffer(gitTarBytes), gitUpload.options);
|
||||
await gitUpload.finish(gitTarBytes.byteLength, gitTarBytes.byteLength);
|
||||
});
|
||||
await gitTimer.time("extract", () => input.client.run(
|
||||
`sh -c ${shellQuote(
|
||||
`mkdir -p ${shellQuote(workspaceRemoteDir)} && ` +
|
||||
`find ${shellQuote(workspaceRemoteDir)} -mindepth 1 -maxdepth 1 ${preserveFindArgs([".paperclip-runtime"])} -exec rm -rf -- {} + && ` +
|
||||
`tar -xf ${shellQuote(remoteGitTar)} -C ${shellQuote(workspaceRemoteDir)} && ` +
|
||||
`rm -f ${shellQuote(remoteGitTar)}`,
|
||||
)}`,
|
||||
{ timeoutMs: input.spec.timeoutMs },
|
||||
));
|
||||
});
|
||||
await gitTimer.finish();
|
||||
} finally {
|
||||
await gitTimer.finish();
|
||||
}
|
||||
}
|
||||
|
||||
const workspaceTarPath = path.join(tempDir, "workspace.tar");
|
||||
@@ -587,52 +590,55 @@ export async function prepareSandboxManagedRuntime(input: {
|
||||
artifact: "workspace",
|
||||
transport: syncTransport,
|
||||
});
|
||||
const workspaceTarBytes = await workspaceTimer.time("tar", async () => {
|
||||
if (gitSnapshot) {
|
||||
await copySelectedWorkspaceEntries({
|
||||
sourceDir: input.workspaceLocalDir,
|
||||
targetDir: workspaceArchiveDir,
|
||||
relativePaths: gitSnapshot.overlayPaths,
|
||||
exclude: workspaceArchiveExclude,
|
||||
try {
|
||||
const workspaceTarBytes = await workspaceTimer.time("tar", async () => {
|
||||
if (gitSnapshot) {
|
||||
await copySelectedWorkspaceEntries({
|
||||
sourceDir: input.workspaceLocalDir,
|
||||
targetDir: workspaceArchiveDir,
|
||||
relativePaths: gitSnapshot.overlayPaths,
|
||||
exclude: workspaceArchiveExclude,
|
||||
});
|
||||
}
|
||||
await createTarballFromDirectory({
|
||||
localDir: workspaceArchiveDir,
|
||||
archivePath: workspaceTarPath,
|
||||
exclude: gitSnapshot ? undefined : workspaceArchiveExclude,
|
||||
});
|
||||
}
|
||||
await createTarballFromDirectory({
|
||||
localDir: workspaceArchiveDir,
|
||||
archivePath: workspaceTarPath,
|
||||
exclude: gitSnapshot ? undefined : workspaceArchiveExclude,
|
||||
return fs.readFile(workspaceTarPath);
|
||||
});
|
||||
return fs.readFile(workspaceTarPath);
|
||||
});
|
||||
const remoteWorkspaceTar = path.posix.join(runtimeRootDir, "workspace-upload.tar");
|
||||
await input.client.makeDir(runtimeRootDir);
|
||||
const workspaceUpload = makeTransferProgress(
|
||||
input.onProgress,
|
||||
"Syncing",
|
||||
"to",
|
||||
"workspace",
|
||||
{ sink: input.onRuntimeProgress, phase: "config_sync" },
|
||||
);
|
||||
await workspaceTimer.time("upload", async () => {
|
||||
await input.client.writeFile(
|
||||
remoteWorkspaceTar,
|
||||
toArrayBuffer(workspaceTarBytes),
|
||||
workspaceUpload.options,
|
||||
const remoteWorkspaceTar = path.posix.join(runtimeRootDir, "workspace-upload.tar");
|
||||
await input.client.makeDir(runtimeRootDir);
|
||||
const workspaceUpload = makeTransferProgress(
|
||||
input.onProgress,
|
||||
"Syncing",
|
||||
"to",
|
||||
"workspace",
|
||||
{ sink: input.onRuntimeProgress, phase: "config_sync" },
|
||||
);
|
||||
await workspaceUpload.finish(workspaceTarBytes.byteLength, workspaceTarBytes.byteLength);
|
||||
});
|
||||
const extractWorkspaceTarCommand = gitSnapshot
|
||||
? `mkdir -p ${shellQuote(workspaceRemoteDir)} && ` +
|
||||
`tar -xf ${shellQuote(remoteWorkspaceTar)} -C ${shellQuote(workspaceRemoteDir)} && ` +
|
||||
`rm -f ${shellQuote(remoteWorkspaceTar)}`
|
||||
: `mkdir -p ${shellQuote(workspaceRemoteDir)} && ` +
|
||||
`find ${shellQuote(workspaceRemoteDir)} -mindepth 1 -maxdepth 1 ${preserveFindArgs([...preservedNames])} -exec rm -rf -- {} + && ` +
|
||||
`tar -xf ${shellQuote(remoteWorkspaceTar)} -C ${shellQuote(workspaceRemoteDir)} && ` +
|
||||
`rm -f ${shellQuote(remoteWorkspaceTar)}`;
|
||||
await workspaceTimer.time("extract", () => input.client.run(
|
||||
`sh -c ${shellQuote(extractWorkspaceTarCommand)}`,
|
||||
{ timeoutMs: input.spec.timeoutMs },
|
||||
));
|
||||
await workspaceTimer.finish();
|
||||
await workspaceTimer.time("upload", async () => {
|
||||
await input.client.writeFile(
|
||||
remoteWorkspaceTar,
|
||||
toArrayBuffer(workspaceTarBytes),
|
||||
workspaceUpload.options,
|
||||
);
|
||||
await workspaceUpload.finish(workspaceTarBytes.byteLength, workspaceTarBytes.byteLength);
|
||||
});
|
||||
const extractWorkspaceTarCommand = gitSnapshot
|
||||
? `mkdir -p ${shellQuote(workspaceRemoteDir)} && ` +
|
||||
`tar -xf ${shellQuote(remoteWorkspaceTar)} -C ${shellQuote(workspaceRemoteDir)} && ` +
|
||||
`rm -f ${shellQuote(remoteWorkspaceTar)}`
|
||||
: `mkdir -p ${shellQuote(workspaceRemoteDir)} && ` +
|
||||
`find ${shellQuote(workspaceRemoteDir)} -mindepth 1 -maxdepth 1 ${preserveFindArgs([...preservedNames])} -exec rm -rf -- {} + && ` +
|
||||
`tar -xf ${shellQuote(remoteWorkspaceTar)} -C ${shellQuote(workspaceRemoteDir)} && ` +
|
||||
`rm -f ${shellQuote(remoteWorkspaceTar)}`;
|
||||
await workspaceTimer.time("extract", () => input.client.run(
|
||||
`sh -c ${shellQuote(extractWorkspaceTarCommand)}`,
|
||||
{ timeoutMs: input.spec.timeoutMs },
|
||||
));
|
||||
} finally {
|
||||
await workspaceTimer.finish();
|
||||
}
|
||||
if (gitSnapshot) {
|
||||
await removeDeletedPathsInSandbox({
|
||||
client: input.client,
|
||||
@@ -651,51 +657,54 @@ export async function prepareSandboxManagedRuntime(input: {
|
||||
transport: syncTransport,
|
||||
});
|
||||
const assetTarPath = path.join(tempDir, `${asset.key}.tar`);
|
||||
const assetTarBytes = await assetTimer.time("tar", async () => {
|
||||
await createTarballFromDirectory({
|
||||
localDir: asset.localDir,
|
||||
archivePath: assetTarPath,
|
||||
followSymlinks: asset.followSymlinks,
|
||||
exclude: asset.exclude,
|
||||
try {
|
||||
const assetTarBytes = await assetTimer.time("tar", async () => {
|
||||
await createTarballFromDirectory({
|
||||
localDir: asset.localDir,
|
||||
archivePath: assetTarPath,
|
||||
followSymlinks: asset.followSymlinks,
|
||||
exclude: asset.exclude,
|
||||
});
|
||||
return fs.readFile(assetTarPath);
|
||||
});
|
||||
return fs.readFile(assetTarPath);
|
||||
});
|
||||
const remoteAssetDir = path.posix.join(runtimeRootDir, asset.key);
|
||||
const remoteAssetTar = path.posix.join(runtimeRootDir, `${asset.key}-upload.tar`);
|
||||
const assetUpload = makeTransferProgress(
|
||||
input.onProgress,
|
||||
"Syncing",
|
||||
"to",
|
||||
asset.key,
|
||||
{ sink: input.onRuntimeProgress, phase: "config_sync" },
|
||||
);
|
||||
await assetTimer.time("upload", async () => {
|
||||
await input.client.writeFile(remoteAssetTar, toArrayBuffer(assetTarBytes), assetUpload.options);
|
||||
await assetUpload.finish(assetTarBytes.byteLength, assetTarBytes.byteLength);
|
||||
for (const stageFile of asset.provision?.stageFiles ?? []) {
|
||||
const stageBytes = typeof stageFile.contents === "string"
|
||||
? Buffer.from(stageFile.contents)
|
||||
: stageFile.contents;
|
||||
const safeName = stageFile.name;
|
||||
if (/[\\/]|\.\.(\.|$)/.test(safeName) || safeName === "..") {
|
||||
throw new Error(`provision stageFile.name must be a simple basename, got: ${safeName}`);
|
||||
const remoteAssetDir = path.posix.join(runtimeRootDir, asset.key);
|
||||
const remoteAssetTar = path.posix.join(runtimeRootDir, `${asset.key}-upload.tar`);
|
||||
const assetUpload = makeTransferProgress(
|
||||
input.onProgress,
|
||||
"Syncing",
|
||||
"to",
|
||||
asset.key,
|
||||
{ sink: input.onRuntimeProgress, phase: "config_sync" },
|
||||
);
|
||||
await assetTimer.time("upload", async () => {
|
||||
await input.client.writeFile(remoteAssetTar, toArrayBuffer(assetTarBytes), assetUpload.options);
|
||||
await assetUpload.finish(assetTarBytes.byteLength, assetTarBytes.byteLength);
|
||||
for (const stageFile of asset.provision?.stageFiles ?? []) {
|
||||
const stageBytes = typeof stageFile.contents === "string"
|
||||
? Buffer.from(stageFile.contents)
|
||||
: stageFile.contents;
|
||||
const safeName = stageFile.name;
|
||||
if (/[\\/]|\.\.(\.|$)/.test(safeName) || safeName === "..") {
|
||||
throw new Error(`provision stageFile.name must be a simple basename, got: ${safeName}`);
|
||||
}
|
||||
await input.client.writeFile(
|
||||
path.posix.join(runtimeRootDir, safeName),
|
||||
toArrayBuffer(stageBytes),
|
||||
);
|
||||
}
|
||||
await input.client.writeFile(
|
||||
path.posix.join(runtimeRootDir, safeName),
|
||||
toArrayBuffer(stageBytes),
|
||||
);
|
||||
}
|
||||
});
|
||||
const extractCommand = asset.provision?.extractCommand?.({
|
||||
assetTarPath: remoteAssetTar,
|
||||
assetDir: remoteAssetDir,
|
||||
runtimeRootDir,
|
||||
}) ?? buildDefaultExtractRuntimeAssetCommand({ remoteAssetDir, remoteAssetTar });
|
||||
await assetTimer.time("extract", () => input.client.run(
|
||||
`sh -c ${shellQuote(extractCommand)}`,
|
||||
{ timeoutMs: input.spec.timeoutMs },
|
||||
));
|
||||
await assetTimer.finish();
|
||||
});
|
||||
const extractCommand = asset.provision?.extractCommand?.({
|
||||
assetTarPath: remoteAssetTar,
|
||||
assetDir: remoteAssetDir,
|
||||
runtimeRootDir,
|
||||
}) ?? buildDefaultExtractRuntimeAssetCommand({ remoteAssetDir, remoteAssetTar });
|
||||
await assetTimer.time("extract", () => input.client.run(
|
||||
`sh -c ${shellQuote(extractCommand)}`,
|
||||
{ timeoutMs: input.spec.timeoutMs },
|
||||
));
|
||||
} finally {
|
||||
await assetTimer.finish();
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
|
||||
Reference in new issue
Block a user