From 17c3be7350960197f787852352d9593720dfa4d8 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 22 May 2026 14:29:30 +0000 Subject: [PATCH] Preserve cancelled cloud upstream runs --- server/src/__tests__/cloud-upstreams.test.ts | 46 +++++++++++++++++++- server/src/services/cloud-upstreams.ts | 37 ++++++++++++---- 2 files changed, 73 insertions(+), 10 deletions(-) diff --git a/server/src/__tests__/cloud-upstreams.test.ts b/server/src/__tests__/cloud-upstreams.test.ts index ca24a5a8456..868669d17e1 100644 --- a/server/src/__tests__/cloud-upstreams.test.ts +++ b/server/src/__tests__/cloud-upstreams.test.ts @@ -1,4 +1,4 @@ -import { randomUUID } from "node:crypto"; +import { generateKeyPairSync, randomUUID } from "node:crypto"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; import { companies, cloudUpstreamConnections, cloudUpstreamRuns, createDb } from "@paperclipai/db"; @@ -222,6 +222,47 @@ describeEmbeddedPostgres("cloud upstream persistence", () => { }); }); + it("preserves a cancelled run when an in-flight createRun tries to finish", async () => { + const companyId = randomUUID(); + const connectionId = randomUUID(); + await seedCompany(companyId); + await db.insert(cloudUpstreamConnections).values(cloudConnectionRow({ id: connectionId, companyId })); + + const service = cloudUpstreamService(db); + const remoteCalls: string[] = []; + globalThis.fetch = vi.fn(async (input) => { + const path = new URL(String(input)).pathname; + remoteCalls.push(path); + if (path.endsWith("/upstream-imports/runs")) { + return jsonResponse({ run: { id: "remote-run-1" } }); + } + if (path.endsWith("/chunks")) { + const run = await db.select().from(cloudUpstreamRuns).then((rows) => rows[0]); + expect(run?.status).toBe("running"); + await service.cancelRun(connectionId, run.id, companyId); + return jsonResponse({ ok: true }); + } + if (path.endsWith("/cancel")) { + return jsonResponse({ ok: true }); + } + if (path.endsWith("/apply")) { + return jsonResponse({ ok: true }); + } + if (path.endsWith("/events")) { + return jsonResponse({ events: [] }); + } + return jsonResponse({ error: "not_found" }, 404); + }) as typeof fetch; + + const result = await service.createRun({ connectionId, companyId }); + + expect(result.status).toBe("cancelled"); + expect(remoteCalls.some((path) => path.endsWith("/apply"))).toBe(false); + const rows = await db.select().from(cloudUpstreamRuns); + expect(rows).toHaveLength(1); + expect(rows[0]?.status).toBe("cancelled"); + }); + async function seedCompany(companyId: string) { await db.insert(companies).values({ id: companyId, @@ -240,6 +281,7 @@ function jsonResponse(body: unknown): Response { } function cloudConnectionRow(input: { id: string; companyId: string }) { + const { privateKey } = generateKeyPairSync("ed25519"); return { id: input.id, companyId: input.companyId, @@ -247,7 +289,7 @@ function cloudConnectionRow(input: { id: string; companyId: string }) { sourceInstanceId: "source-1", sourceInstanceFingerprint: "sha256:test", sourcePublicKey: "public-key", - privateKeyPem: "legacy-private-key", + privateKeyPem: privateKey.export({ type: "pkcs8", format: "pem" }).toString(), tokenStatus: "connected", scopes: ["upstream_import:write"], authorizedGlobalUserId: "user-1", diff --git a/server/src/services/cloud-upstreams.ts b/server/src/services/cloud-upstreams.ts index f5330ba369a..765dd4329b3 100644 --- a/server/src/services/cloud-upstreams.ts +++ b/server/src/services/cloud-upstreams.ts @@ -332,7 +332,7 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } = entities: bundle.entities, }); const remoteRunId = remoteRunIdFromResponse(remoteRun); - await updateRun(runId, { + const pushedRun = await updateRunIfRunning(runId, { remoteRunId, activeStep: "push", progressPercent: 60, @@ -341,11 +341,12 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } = event(new Date().toISOString(), "push", "updated", "Created or resumed the cloud import ledger run."), ], }); + if (pushedRun.status !== "running") return pushedRun; for (const chunk of bundle.chunks) { await remotePost(connection, `/api/upstream-import-runs/${encodeURIComponent(remoteRunId)}/chunks`, chunk); } - await updateRun(runId, { + const verifiedRun = await updateRunIfRunning(runId, { activeStep: "verify", progressPercent: 82, events: [ @@ -353,6 +354,7 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } = event(new Date().toISOString(), "push", "completed", `Uploaded ${bundle.chunks.length} manifest chunk${bundle.chunks.length === 1 ? "" : "s"}.`), ], }); + if (verifiedRun.status !== "running") return verifiedRun; const applied = await remotePost(connection, `/api/upstream-import-runs/${encodeURIComponent(remoteRunId)}/apply`, {}); const remoteEvents = await remoteGet(connection, `/api/upstream-import-runs/${encodeURIComponent(remoteRunId)}/events`).catch(() => null); @@ -364,7 +366,7 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } = event(completedAt.toISOString(), "activate", "completed", "Activation checklist is ready for manual unpause decisions."), ...eventsFromRemote(remoteEvents), ]; - const finalRun = await updateRun(runId, { + const finalRun = await updateRunIfRunning(runId, { remoteRunId, status: "succeeded", activeStep: "activate", @@ -384,15 +386,17 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } = }, completedAt, }); - await db - .update(cloudUpstreamConnections) - .set({ lastRunId: finalRun.id, updatedAt: new Date() }) - .where(eq(cloudUpstreamConnections.id, connection.id)); + if (finalRun.status === "succeeded") { + await db + .update(cloudUpstreamConnections) + .set({ lastRunId: finalRun.id, updatedAt: new Date() }) + .where(eq(cloudUpstreamConnections.id, connection.id)); + } return finalRun; } catch (error) { const failedAt = new Date(); const failure = cloudUpstreamRemoteFailureReport(error); - return updateRun(runId, { + return updateRunIfRunning(runId, { status: "failed", activeStep: "push", progressPercent: 100, @@ -537,6 +541,23 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } = return runFromRow(updated); } + async function updateRunIfRunning(runId: string, patch: Partial): Promise { + const [updated] = await db + .update(cloudUpstreamRuns) + .set({ ...patch, updatedAt: new Date() }) + .where(and(eq(cloudUpstreamRuns.id, runId), eq(cloudUpstreamRuns.status, "running"))) + .returning(); + if (updated) return runFromRow(updated); + + const [current] = await db + .select() + .from(cloudUpstreamRuns) + .where(eq(cloudUpstreamRuns.id, runId)) + .limit(1); + if (!current) throw notFound("Cloud upstream run was not found"); + return runFromRow(current); + } + async function localPreview(connection: ConnectionRow): Promise { return { connectionId: connection.id,