diff --git a/server/src/__tests__/remote-snapshot-ledger-settlement.test.ts b/server/src/__tests__/remote-snapshot-ledger-settlement.test.ts index 6b532613118..e4df204ce36 100644 --- a/server/src/__tests__/remote-snapshot-ledger-settlement.test.ts +++ b/server/src/__tests__/remote-snapshot-ledger-settlement.test.ts @@ -1,5 +1,5 @@ import { randomUUID } from "node:crypto"; -import { afterAll, beforeAll, beforeEach, describe, expect, it } from "vitest"; +import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { eq, sql } from "drizzle-orm"; import { agents, @@ -19,6 +19,11 @@ import { } from "../services/recovery/ownership-invariant-report.js"; import { OWNABLE_KINDS } from "../services/recovery/ownership-registry.js"; +const aborted = vi.hoisted(() => ({ keys: [] as string[] })); +vi.mock("../storage/index.js", () => ({ + getStorageService: () => ({ abortIncompleteUploads: async (_companyId: string, key: string) => { aborted.keys.push(key); return 1; } }), +})); + const external = resolveExternalTestDatabaseUrl("PAPERCLIP_HISTORY_TEST_DATABASE_URL"); const minutesAgo = (minutes: number) => new Date(Date.now() - minutes * 60_000); const kindReport = (report: OwnershipReport, kind: string) => report.kinds.find(entry => entry.kind === kind)!; @@ -44,7 +49,7 @@ describe("remote snapshot ledger settlement", () => { beforeEach(async () => { await db.execute(sql`truncate execution_workspace_snapshots, activity_log, workspace_operations, execution_workspaces, heartbeat_runs, agents, companies cascade`); - companyId = randomUUID(); agentId = randomUUID(); + companyId = randomUUID(); agentId = randomUUID(); aborted.keys = []; await db.insert(companies).values({ id: companyId, name: "Ledger fixture", issuePrefix: "LED" }); await db.insert(agents).values({ id: agentId, companyId, name: "Fixture", role: "engineer", adapterType: "codex_local" }); }); @@ -140,4 +145,25 @@ describe("remote snapshot ledger settlement", () => { expect(await ledger(snapshotId)).toMatchObject({ status: "deleted", metadata: { abandonedUpload: { runId: failed, operationId } } }); }); + + it("aborts the incomplete multipart uploads of every attempt when the reconciler retires an abandoned save", async () => { + const failed = await run("failed"); + const ledgerWorkspace = await workspace({ epoch: 2, writerRunId: null, dirty: false }); + const ledgerId = randomUUID(), ledgerKeys = ["a", "b"].map(name => `${companyId}/workspace-snapshots/${ledgerWorkspace}/${ledgerId}/${name}.pcws`); + await save({ workspaceId: ledgerWorkspace, runId: failed, + snapshot: { id: ledgerId, metadata: { uploads: [...ledgerKeys.map(objectKey => ({ objectKey })), { objectKey: `${companyId}/elsewhere/c.pcws` }] } } }); + await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), + kinds: OWNABLE_KINDS.filter(kind => kind.id === "remote_snapshot_ledger") }); + expect(aborted.keys).toEqual(ledgerKeys); + + aborted.keys = []; + const saveWorkspace = await workspace({ epoch: 2, writerRunId: null, dirty: false }); + const saveId = randomUUID(), saveKey = `${companyId}/workspace-snapshots/${saveWorkspace}/${saveId}/cut.pcws`; + await save({ workspaceId: saveWorkspace, runId: failed, status: "running", lastFailure: false, + snapshot: { id: saveId, metadata: { uploads: [{ objectKey: saveKey }] } } }); + await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), + kinds: OWNABLE_KINDS.filter(kind => kind.id === "remote_snapshot_save") }); + expect(await ledger(saveId)).toMatchObject({ status: "delete_requested" }); + expect(aborted.keys).toEqual([saveKey]); + }); }); diff --git a/server/src/__tests__/snapshot-attempt-upload-abort.test.ts b/server/src/__tests__/snapshot-attempt-upload-abort.test.ts new file mode 100644 index 00000000000..775b8873297 --- /dev/null +++ b/server/src/__tests__/snapshot-attempt-upload-abort.test.ts @@ -0,0 +1,33 @@ +import { describe, expect, it } from "vitest"; +import { STORAGE_FAILURE_CLEANUP_MS } from "../storage/failure-cleanup.js"; +import type { StorageService } from "../storage/types.js"; +import { abortSnapshotAttemptUploads } from "../services/remote-workspace-snapshot-retention.js"; + +describe("aborting a snapshot's attempt uploads", () => { + it("shares one deadline across every attempt key of a hanging store", async () => { + const started: string[] = []; + const storage = { + abortIncompleteUploads: (_companyId: string, key: string, options?: { signal?: AbortSignal }) => { + started.push(key); + return new Promise((_resolve, reject) => { + options?.signal?.addEventListener("abort", () => reject(options.signal!.reason), { once: true }); + }); + }, + } as unknown as StorageService; + const keys = Array.from({ length: 20 }, (_, index) => `company/workspace-snapshots/ws/s/${index}.pcws`); + const begun = Date.now(); + expect(await abortSnapshotAttemptUploads(storage, "company", keys)).toBe(0); + const elapsed = Date.now() - begun; + expect(elapsed).toBeGreaterThanOrEqual(STORAGE_FAILURE_CLEANUP_MS - 100); + expect(elapsed).toBeLessThan(2 * STORAGE_FAILURE_CLEANUP_MS); + expect(started).toHaveLength(8); + }, 3 * STORAGE_FAILURE_CLEANUP_MS); + + it("aborts every key of a responsive store and counts each abort", async () => { + const seen: string[] = []; + const storage = { abortIncompleteUploads: async (_companyId: string, key: string) => { seen.push(key); return 1; } } as unknown as StorageService; + const keys = Array.from({ length: 20 }, (_, index) => `k${index}`); + expect(await abortSnapshotAttemptUploads(storage, "company", keys)).toBe(20); + expect(seen.sort()).toEqual([...keys].sort()); + }); +}); diff --git a/server/src/__tests__/storage-s3-multipart-shutdown.test.ts b/server/src/__tests__/storage-s3-multipart-shutdown.test.ts new file mode 100644 index 00000000000..54f9065828e --- /dev/null +++ b/server/src/__tests__/storage-s3-multipart-shutdown.test.ts @@ -0,0 +1,151 @@ +import { AbortMultipartUploadCommand, CompleteMultipartUploadCommand, CreateMultipartUploadCommand, ListMultipartUploadsCommand, S3Client, UploadPartCommand } from "@aws-sdk/client-s3"; +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { Readable } from "node:stream"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { abortInFlightMultipartUploads, putS3Multipart, putS3MultipartFile } from "../storage/s3-multipart.js"; +import { createS3StorageProvider } from "../storage/s3-provider.js"; +import { afterStalledShutdownStep, runBoundedShutdownStep } from "../shutdown.js"; + +const MIB = 1024 * 1024; + +function heldPartClient() { + const client = new S3Client({ region: "us-east-1" }), aborted: string[] = []; + let partStarted!: () => void; + const started = new Promise(resolve => { partStarted = resolve; }); + vi.spyOn(client, "send").mockImplementation(async (command: any, options?: any) => { + if (command instanceof CreateMultipartUploadCommand) return { UploadId: "held-upload" }; + if (command instanceof UploadPartCommand) { + partStarted(); + const signal = options?.abortSignal as AbortSignal; + return new Promise((_resolve, reject) => { + if (signal.aborted) reject(signal.reason); + signal.addEventListener("abort", () => reject(signal.reason), { once: true }); + }); + } + if (command instanceof AbortMultipartUploadCommand) { aborted.push(command.input.UploadId!); return {}; } + throw new Error(`unexpected ${command.constructor.name}`); + }); + return { client, aborted, started }; +} + +describe("multipart uploads in flight at shutdown", () => { + let root: string | null = null; + afterEach(async () => { vi.restoreAllMocks(); if (root) await fs.rm(root, { recursive: true, force: true }); root = null; }); + + it("has nothing to abort when no upload is running", async () => { + expect(await abortInFlightMultipartUploads(new Error("server_shutdown"))).toBe(0); + }); + + it("aborts a file upload whose part the store is holding", async () => { + root = await fs.mkdtemp(path.join(os.tmpdir(), "pc-multipart-shutdown-")); + const file = path.join(root, "archive.pcws"); + await fs.writeFile(file, Buffer.alloc(17 * MIB, 7)); + const { client, aborted, started } = heldPartClient(); + const saving = putS3MultipartFile(client, "bucket", "key", { objectKey: "key", contentType: "application/octet-stream", + contentLength: 17 * MIB, body: Readable.from([]), sourceFile: file }).catch(error => error); + await started; + expect(await abortInFlightMultipartUploads(new Error("server_shutdown"))).toBe(1); + expect(aborted).toEqual(["held-upload"]); + expect(((await saving) as Error).message).toBe("server_shutdown"); + expect(await abortInFlightMultipartUploads(new Error("server_shutdown"))).toBe(0); + }); + + it("lets an upload that finishes inside the grace complete without aborting it", async () => { + root = await fs.mkdtemp(path.join(os.tmpdir(), "pc-multipart-shutdown-")); + const file = path.join(root, "archive.pcws"); + await fs.writeFile(file, Buffer.alloc(17 * MIB, 7)); + const client = new S3Client({ region: "us-east-1" }), sent: string[] = []; + let release!: () => void, partStarted!: () => void; + const released = new Promise(resolve => { release = resolve; }); + const started = new Promise(resolve => { partStarted = resolve; }); + vi.spyOn(client, "send").mockImplementation(async (command: any) => { + sent.push(command.constructor.name); + if (command instanceof CreateMultipartUploadCommand) return { UploadId: "graceful-upload" }; + if (command instanceof UploadPartCommand) { partStarted(); await released; return { ETag: `part-${command.input.PartNumber}` }; } + if (command instanceof CompleteMultipartUploadCommand) return {}; + throw new Error(`unexpected ${command.constructor.name}`); + }); + const saving = putS3MultipartFile(client, "bucket", "key", { objectKey: "key", contentType: "application/octet-stream", + contentLength: 17 * MIB, body: Readable.from([]), sourceFile: file }); + await started; + const draining = abortInFlightMultipartUploads(new Error("server_shutdown"), { graceMs: 5_000 }); + setTimeout(release, 50); + expect(await draining).toBe(0); + expect(await saving).toEqual({ integrity: "server_verified_md5" }); + expect(sent).toContain("CompleteMultipartUploadCommand"); + expect(sent).not.toContain("AbortMultipartUploadCommand"); + }); + + it("aborts an upload still held after the grace and counts it once", async () => { + root = await fs.mkdtemp(path.join(os.tmpdir(), "pc-multipart-shutdown-")); + const file = path.join(root, "archive.pcws"); + await fs.writeFile(file, Buffer.alloc(17 * MIB, 7)); + const { client, aborted, started } = heldPartClient(); + const saving = putS3MultipartFile(client, "bucket", "key", { objectKey: "key", contentType: "application/octet-stream", + contentLength: 17 * MIB, body: Readable.from([]), sourceFile: file }).catch(error => error); + await started; + const begun = Date.now(); + expect(await abortInFlightMultipartUploads(new Error("server_shutdown"), { graceMs: 100 })).toBe(1); + expect(Date.now() - begun).toBeGreaterThanOrEqual(90); + expect(aborted).toEqual(["held-upload"]); + expect(((await saving) as Error).message).toBe("server_shutdown"); + }); + + it("aborts a streamed upload whose part the store is holding", async () => { + const { client, aborted, started } = heldPartClient(); + const body = Readable.from((async function* () { for (let index = 0; index < 272; index++) yield Buffer.alloc(64 * 1024, 3); })()); + const saving = putS3Multipart(client, "bucket", "key", { objectKey: "key", contentType: "application/octet-stream", + contentLength: 17 * MIB, body }).catch(error => error); + await started; + expect(await abortInFlightMultipartUploads(new Error("server_shutdown"))).toBe(1); + expect(aborted).toEqual(["held-upload"]); + expect(((await saving) as Error).message).toBe("server_shutdown"); + }); +}); + +describe("aborting a key's incomplete multipart uploads", () => { + afterEach(() => { vi.restoreAllMocks(); vi.unstubAllEnvs(); }); + + it("aborts every upload of the exact key across listing pages and leaves neighboring keys alone", async () => { + vi.stubEnv("AWS_ACCESS_KEY_ID", "synthetic-access-key"); + vi.stubEnv("AWS_SECRET_ACCESS_KEY", "synthetic-secret-key"); + const aborted: string[] = [], listed: Array<{ Prefix?: string; KeyMarker?: string; UploadIdMarker?: string }> = []; + const send = vi.spyOn(S3Client.prototype, "send").mockImplementation(async (command: any) => { + if (command instanceof ListMultipartUploadsCommand) { + listed.push(command.input); + if (!command.input.KeyMarker) return { IsTruncated: true, NextKeyMarker: "p/c/a.pcws", NextUploadIdMarker: "u2", + Uploads: [{ Key: "p/c/a.pcws", UploadId: "u1" }, { Key: "p/c/a.pcws", UploadId: "u2" }] }; + return { IsTruncated: false, Uploads: [{ Key: "p/c/a.pcws.next", UploadId: "u3" }, { Key: "p/c/a.pcws", UploadId: "u4" }] }; + } + if (command instanceof AbortMultipartUploadCommand) { + if (command.input.UploadId === "u4") throw Object.assign(new Error("gone"), { name: "NoSuchUpload" }); + aborted.push(`${command.input.Key}#${command.input.UploadId}`); return {}; + } + throw new Error(`unexpected ${command.constructor.name}`); + }); + const provider = createS3StorageProvider({ bucket: "bucket", region: "us-east-1", endpoint: "http://127.0.0.1:9", prefix: "p" }); + expect(await provider.abortIncompleteUploads!({ objectKey: "c/a.pcws" })).toBe(2); + expect(listed.map(input => [input.Prefix, input.KeyMarker, input.UploadIdMarker])) + .toEqual([["p/c/a.pcws", undefined, undefined], ["p/c/a.pcws", "p/c/a.pcws", "u2"]]); + expect(aborted).toEqual(["p/c/a.pcws#u1", "p/c/a.pcws#u2"]); + send.mockRestore(); + }); +}); + +describe("shutdown steps that stall on a held upload", () => { + const log = { info: () => {}, error: () => {} }; + + it("aborts in-flight uploads after a step that timed out or failed, and not after one that completed", async () => { + const onStalled = vi.fn(async () => {}); + const held = new Promise(() => {}); + await afterStalledShutdownStep(await runBoundedShutdownStep({ step: "workspace_recovery_stop", signal: "SIGTERM", timeoutMs: 5, run: () => held, log }), onStalled); + expect(onStalled).toHaveBeenCalledTimes(1); + await afterStalledShutdownStep(await runBoundedShutdownStep({ step: "workspace_recovery_stop", signal: "SIGTERM", timeoutMs: 5, + run: async () => { throw new Error("stop failed"); }, log }), onStalled); + expect(onStalled).toHaveBeenCalledTimes(2); + await afterStalledShutdownStep(await runBoundedShutdownStep({ step: "workspace_recovery_stop", signal: "SIGTERM", timeoutMs: 1000, run: async () => {}, log }), onStalled); + expect(onStalled).toHaveBeenCalledTimes(2); + }); +}); diff --git a/server/src/index.ts b/server/src/index.ts index 7f13f2f2dd1..8e785370055 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -195,6 +195,7 @@ import { buildRuntimeApiCandidateUrls, choosePrimaryRuntimeApiUrl, resolveLocalR import { isLoopbackHost, rewriteLoopbackUrlPort } from "./url-utils.js"; import { createPluginWorkerManager } from "./services/plugin-worker-manager.js"; import { createStorageServiceFromConfig } from "./storage/index.js"; +import { abortInFlightMultipartUploads } from "./storage/s3-multipart.js"; import { printStartupBanner } from "./startup-banner.js"; import { getBoardClaimWarningUrl, initializeBoardClaimChallenge } from "./board-claim.js"; import { maybePersistWorktreeRuntimePorts } from "./worktree-config.js"; @@ -203,6 +204,7 @@ import { conflict } from "./errors.js"; import { ensureDecisionSigningSecret } from "./services/decision-signing.js"; import { createDecisionRetentionNotifyOriginAgent, createDecisionWakeOriginAgent } from "./services/decision-wakeup.js"; import { + afterStalledShutdownStep, closeHttpListenerForShutdown, coordinateHeartbeatSchedulerShutdown, createPreDrainShutdownBudget, @@ -2044,6 +2046,11 @@ async function startServerWithDatabaseTeardown( ); } + const abortStuckMultipartUploads = async (signal: "SIGINT" | "SIGTERM", graceMs = 0) => { + const abortedUploads = await abortInFlightMultipartUploads(new Error("server_shutdown"), { graceMs }); + if (abortedUploads) logger.info({ signal, abortedUploads }, "aborted in-flight multipart uploads before exit"); + }; + const performShutdown = async ( signal: "SIGINT" | "SIGTERM", exitProcess: boolean, @@ -2071,10 +2078,11 @@ async function startServerWithDatabaseTeardown( unsubscribeRetainedNativeSessionCleanup?.(); // B can accept Stop/answers while A still owns an execution. Preserve the // owner command workers and their transport until the handoff below ends. - await preDrain.run("delivery_work_stop", () => deliveryWork.stop(deploymentHints ? new Set([ + const deliveryWorkStop = await preDrain.run("delivery_work_stop", () => deliveryWork.stop(deploymentHints ? new Set([ DELIVERY_QUEUES.nativeCancellation, DELIVERY_QUEUES.nativeRuntimeResponses, DELIVERY_QUEUES.question, DELIVERY_QUEUES.deploymentNotifications, DELIVERY_QUEUES.liveInvalidations, ]) : undefined)); + await afterStalledShutdownStep(deliveryWorkStop, () => abortStuckMultipartUploads(signal)); await preDrain.run("run_evidence_migration_stop", async () => { await runEvidenceMigrationWork?.stop(); }); @@ -2083,7 +2091,8 @@ async function startServerWithDatabaseTeardown( await skillVersionRetentionWorker?.dispose(); }); const workspaceRecoveryStop = workspaceRecoveryWork.stop(); - await preDrain.run("workspace_recovery_stop", () => workspaceRecoveryStop); + await afterStalledShutdownStep(await preDrain.run("workspace_recovery_stop", () => workspaceRecoveryStop), + () => abortStuckMultipartUploads(signal)); unregisterTerminalWorkspaceProbe?.(); unsubscribeNativeWorkspaceSyncRetention?.(); @@ -2134,6 +2143,8 @@ async function startServerWithDatabaseTeardown( } } + await abortStuckMultipartUploads(signal, 5_000); + if (!skipHeartbeatDrain) { await drainRunExecutionFinalizersForShutdown({ signal, diff --git a/server/src/services/recovery/remote-snapshot-ledger-settlement.ts b/server/src/services/recovery/remote-snapshot-ledger-settlement.ts index f9e8b26bd27..9a5dbe2bf94 100644 --- a/server/src/services/recovery/remote-snapshot-ledger-settlement.ts +++ b/server/src/services/recovery/remote-snapshot-ledger-settlement.ts @@ -4,7 +4,9 @@ import { readRemoteWorkspaceAuthority } from "@paperclipai/shared"; import { persistActivity, publishActivity, type ActivityPublication } from "../activity-log.js"; import type { OwnerProof } from "../owner-proof.js"; import { readSaveWork } from "../remote-workspaces.js"; -import { notifyRemoteWorkspaceSnapshots } from "../remote-workspace-snapshot-retention.js"; +import { abortSnapshotAttemptUploads, notifyRemoteWorkspaceSnapshots, snapshotAttemptKeys } from "../remote-workspace-snapshot-retention.js"; +import { getStorageService } from "../../storage/index.js"; +import type { StorageService } from "../../storage/types.js"; import type { InProgressRecord, SettleOutcome } from "./ownership-registry.js"; export const ABANDONED_REMOTE_SNAPSHOT_LEDGER_REASON = "abandoned_remote_snapshot_upload"; @@ -55,11 +57,20 @@ export async function retireEndedRemoteSnapshotLedgerRow(tx: Db, input: { return { retired: true, snapshot: row, status }; } +export async function abortRetiredSnapshotUploads(snapshots: Snapshot[], storage?: StorageService) { + const attempted = snapshots.filter(snapshot => snapshotAttemptKeys(snapshot).length); + if (!attempted.length) return; + let service: StorageService; + try { service = storage ?? getStorageService(); } catch { return; } + for (const snapshot of attempted) await abortSnapshotAttemptUploads(service, snapshot.companyId, snapshotAttemptKeys(snapshot)); +} + export async function settleAbandonedRemoteSnapshotLedger(db: Db, record: InProgressRecord, - context: { now: Date; proof: OwnerProof }): Promise { + context: { now: Date; proof: OwnerProof }, options: { storage?: StorageService } = {}): Promise { if (record.owner.kind !== "run" || !record.companyId) return { settled: false, reason: "owner_not_provable" }; const companyId = record.companyId; let publication: ActivityPublication | undefined; + let retired: Snapshot | undefined; const outcome = await db.transaction(async (transaction): Promise => { const tx = transaction as unknown as Db; const result = await retireEndedRemoteSnapshotLedgerRow(tx, { companyId, snapshotId: record.recordId, @@ -75,9 +86,11 @@ export async function settleAbandonedRemoteSnapshotLedger(db: Db, record: InProg ...(result.snapshot.metadata?.abandonedUpload as Record) }, }); publication = activity.publication; + retired = result.snapshot; return { settled: true, reason: context.proof.reason }; }); if (outcome.settled && publication) publishActivity(publication); + if (outcome.settled && retired) await abortRetiredSnapshotUploads([retired], options.storage); return outcome; } @@ -87,7 +100,10 @@ export async function retireAbandonedSaveLedgerRows(tx: Db, input: { const rows = await tx.select({ id: executionWorkspaceSnapshots.id }).from(executionWorkspaceSnapshots).where(and( eq(executionWorkspaceSnapshots.companyId, input.companyId), eq(executionWorkspaceSnapshots.sourceOperationId, input.operationId), eq(executionWorkspaceSnapshots.status, "pending"))); + const retired: Snapshot[] = []; for (const row of rows) { - await retireEndedRemoteSnapshotLedgerRow(tx, { companyId: input.companyId, snapshotId: row.id, now: input.now, proof: input.proof }); + const result = await retireEndedRemoteSnapshotLedgerRow(tx, { companyId: input.companyId, snapshotId: row.id, now: input.now, proof: input.proof }); + if (result.retired) retired.push(result.snapshot); } + return retired; } diff --git a/server/src/services/recovery/workspace-finalize-attempt-settlement.ts b/server/src/services/recovery/workspace-finalize-attempt-settlement.ts index 5fcb5d945fd..1fbe10195e2 100644 --- a/server/src/services/recovery/workspace-finalize-attempt-settlement.ts +++ b/server/src/services/recovery/workspace-finalize-attempt-settlement.ts @@ -9,7 +9,8 @@ import { notifyTerminalWorkspaceWork } from "../terminal-workspace-work.js"; import { RETAINED_SOURCE_FALLBACK_CHECK_KEY } from "../workspace-recovery-fallback-discovery.js"; import { LOCAL_WORKSPACE_COMMAND_PHASES, isOwnerStampedWorkspaceOperationLive } from "../workspace-operations.js"; import type { InProgressRecord, SettleOutcome } from "./ownership-registry.js"; -import { retireAbandonedSaveLedgerRows } from "./remote-snapshot-ledger-settlement.js"; +import type { StorageService } from "../../storage/types.js"; +import { abortRetiredSnapshotUploads, retireAbandonedSaveLedgerRows } from "./remote-snapshot-ledger-settlement.js"; export const ABANDONED_WORKSPACE_FINALIZE_REASON = "abandoned_workspace_finalization"; export const ABANDONED_WORKSPACE_COMMAND_REASON = "abandoned_workspace_command"; @@ -112,8 +113,9 @@ function abandonedWorkspaceOperationSettle(spec: { phases: readonly string[]; st export const ABANDONED_REMOTE_SNAPSHOT_SAVE_REASON = "abandoned_remote_snapshot_save"; export async function settleAbandonedRemoteSnapshotSave(db: Db, record: InProgressRecord, - context: { now: Date; proof: OwnerProof }): Promise { + context: { now: Date; proof: OwnerProof }, options: { storage?: StorageService } = {}): Promise { let publication: ActivityPublication | undefined; + let retired: Awaited> = []; const outcome = await db.transaction(async (transaction): Promise => { const tx = transaction as unknown as Db; const [operation] = await tx.select().from(workspaceOperations) @@ -165,7 +167,7 @@ export async function settleAbandonedRemoteSnapshotSave(db: Db, record: InProgre }).where(and(eq(workspaceOperations.id, operation.id), eq(workspaceOperations.status, operation.status))) .returning({ id: workspaceOperations.id }); if (!row) return { settled: false, reason: "attempt_no_longer_in_progress" }; - await retireAbandonedSaveLedgerRows(tx, { companyId: operation.companyId, operationId: operation.id, now, proof: context.proof.reason }); + retired = await retireAbandonedSaveLedgerRows(tx, { companyId: operation.companyId, operationId: operation.id, now, proof: context.proof.reason }); if (operation.executionWorkspaceId) { await notifyDeliveryWork(tx, DELIVERY_QUEUES.workspaceRecovery); await tx.update(issueRecoveryActions).set({ @@ -185,5 +187,6 @@ export async function settleAbandonedRemoteSnapshotSave(db: Db, record: InProgre return { settled: true, reason: context.proof.reason }; }); if (outcome.settled && publication) publishActivity(publication); + if (outcome.settled) await abortRetiredSnapshotUploads(retired, options.storage); return outcome; } diff --git a/server/src/services/remote-workspace-snapshot-retention.test.ts b/server/src/services/remote-workspace-snapshot-retention.test.ts index 60986fff325..88612a5583a 100644 --- a/server/src/services/remote-workspace-snapshot-retention.test.ts +++ b/server/src/services/remote-workspace-snapshot-retention.test.ts @@ -44,7 +44,10 @@ async function versionedS3Fixture() { const versionId = parsed.searchParams.get("versionId"); const keyMarker = parsed.searchParams.get("key-marker"), versionMarker = parsed.searchParams.get("version-id-marker"); requests.push({ method: req.method!, key, versionId, keyMarker, versionMarker }); - if (parsed.searchParams.has("versions")) { + if (parsed.searchParams.has("uploads")) { + res.setHeader("content-type", "application/xml"); + res.end(`bucketfalse`); + } else if (parsed.searchParams.has("versions")) { if (faults.listDenied) { res.writeHead(403, { "content-type": "application/xml" }); res.end("AccessDeniedVersion listing permission is required"); @@ -142,6 +145,19 @@ describe.skipIf(!support.supported)("remote workspace snapshot retention", () => expect((await db.select().from(executionWorkspaceSnapshots).where(eq(executionWorkspaceSnapshots.id, held.row.id)))[0]!.status).toBe("pending"); for (const key of held.keys) expect((await storage.headObject(held.companyId, key)).exists).toBe(true); }); + it("aborts each attempt's incomplete multipart upload before deleting a retired snapshot's objects", async () => { + const f = await seed(), order: string[] = [], deleteObject = storage.deleteObject.bind(storage); + storage.abortIncompleteUploads = vi.fn(async (_companyId: string, key: string) => { order.push(`abort:${key}`); return 1; }); + storage.deleteObject = async (companyId, key, options) => { order.push(`delete:${key}`); return deleteObject(companyId, key, options); }; + expect(await service().tick()).toMatchObject({ deleted: 1, failed: 0 }); + expect(order).toEqual([...f.keys.map(key => `abort:${key}`), ...f.keys.map(key => `delete:${key}`)]); + }); + it("still retires a snapshot when the store refuses to list its multipart uploads", async () => { + const f = await seed(); + storage.abortIncompleteUploads = vi.fn(async () => { throw Object.assign(new Error("Access Denied"), { name: "AccessDenied" }); }); + expect(await service().tick()).toMatchObject({ deleted: 1, failed: 0 }); + for (const key of f.keys) expect((await storage.headObject(f.companyId, key)).exists).toBe(false); + }); it("deletes every recorded attempt and winning object after grace", async () => { const f = await seed(); const owner = service(); expect(await owner.tick()).toMatchObject({ deleted: 1, failed: 0 }); diff --git a/server/src/services/remote-workspace-snapshot-retention.ts b/server/src/services/remote-workspace-snapshot-retention.ts index 2003dce614e..904558671c2 100644 --- a/server/src/services/remote-workspace-snapshot-retention.ts +++ b/server/src/services/remote-workspace-snapshot-retention.ts @@ -5,9 +5,12 @@ import { hasRemoteTerminationReceipt } from "./remote-execution-termination.js"; import type { StorageService } from "../storage/types.js"; import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; +import { STORAGE_FAILURE_CLEANUP_MS } from "../storage/failure-cleanup.js"; +import { logger } from "../middleware/logger.js"; export const REMOTE_WORKSPACE_SNAPSHOT_GRACE_MS = 7 * 86400_000; export const REMOTE_WORKSPACE_SNAPSHOT_RETRY_MS = 15 * 60_000; +const SNAPSHOT_ABORT_CONCURRENCY = 8; type Snapshot = typeof executionWorkspaceSnapshots.$inferSelect; type Tx = Parameters[0]>[0]; type ReadDb = Pick | Pick; @@ -17,6 +20,38 @@ export function notifyRemoteWorkspaceSnapshots(transaction: object, hint?: strin return notifyDeliveryWork(transaction, DELIVERY_QUEUES.remoteWorkspaceSnapshots, hint); } +export function snapshotAttemptKeys(row: Snapshot) { + const uploads = Array.isArray(row.metadata?.uploads) ? row.metadata.uploads as Array<{ objectKey?: unknown }> : []; + const prefix = `${row.companyId}/workspace-snapshots/${row.executionWorkspaceId}/${row.id}/`; + return [...new Set(uploads.flatMap(upload => typeof upload?.objectKey === "string" && upload.objectKey !== row.objectKey + && upload.objectKey.startsWith(prefix) && !upload.objectKey.includes("..") ? [upload.objectKey] : []))]; +} + +export async function abortSnapshotAttemptUploads(storage: StorageService, companyId: string, keys: Iterable, signal?: AbortSignal) { + const abort = storage.abortIncompleteUploads?.bind(storage); + const pending = [...keys]; + if (!abort || !pending.length) return 0; + signal?.throwIfAborted(); + const bounded = AbortSignal.timeout(STORAGE_FAILURE_CLEANUP_MS); + const deadline = signal ? AbortSignal.any([signal, bounded]) : bounded; + let aborted = 0; + const worker = async () => { + for (let key = pending.shift(); key !== undefined && !deadline.aborted; key = pending.shift()) { + try { + const count = await abort(companyId, key, { signal: deadline }); + aborted += count; + } catch (error) { + if (signal?.aborted) throw error; + logger.warn({ err: error, companyId, objectKey: key }, "workspace snapshot multipart upload abort failed"); + } + } + }; + await Promise.all(Array.from({ length: Math.min(SNAPSHOT_ABORT_CONCURRENCY, pending.length) }, worker)); + signal?.throwIfAborted(); + if (pending.length) logger.warn({ companyId, remaining: pending.length }, "workspace snapshot multipart upload abort ran out of time"); + return aborted; +} + function abandonedUploadOperationId(row: Snapshot) { if (row.status !== "delete_requested") return null; const abandoned = row.metadata?.abandonedUpload as { operationId?: unknown } | undefined; @@ -148,6 +183,7 @@ export function remoteWorkspaceSnapshotRetentionService(db: Db, options: { for (const key of keys) { if (!key.startsWith(prefix) || key.includes("..") || key.includes("\\")) throw new Error("snapshot_upload_scope_invalid"); } + await abortSnapshotAttemptUploads(options.storage, claimed.companyId, keys, signal); for (const key of keys) { signal?.throwIfAborted(); await options.storage.deleteObject(claimed.companyId, key, { signal, allVersions: true }); diff --git a/server/src/services/remote-workspace-snapshot-storage.test.ts b/server/src/services/remote-workspace-snapshot-storage.test.ts index 688b7a25431..d3eab02d39a 100644 --- a/server/src/services/remote-workspace-snapshot-storage.test.ts +++ b/server/src/services/remote-workspace-snapshot-storage.test.ts @@ -244,6 +244,41 @@ suite("remote workspace snapshot storage", () => { const [row] = await db.select().from(executionWorkspaceSnapshots).where(eq(executionWorkspaceSnapshots.id, ref.id)); expect((row?.metadata?.uploads as unknown[]).length).toBe(1); }); + it("aborts the incomplete upload of an attempt a restart cut off once a retried save publishes", async () => { + const f = await fixture(), put = storage.putFile.bind(storage), incomplete = new Set(); + storage.abortIncompleteUploads = vi.fn(async (_companyId: string, key: string) => Number(incomplete.delete(key))); + storage.putFile = async input => { incomplete.add(input.objectKey!); throw new Error("process killed mid UploadPart"); }; + await expect(uploadWorkspaceSnapshot(f.upload)).rejects.toThrow("process killed mid UploadPart"); + const [cut] = [...incomplete]; + storage.putFile = put; + const ref = await uploadWorkspaceSnapshot(f.upload); + expect(ref.objectKey).not.toBe(cut); + expect(storage.abortIncompleteUploads).toHaveBeenCalledTimes(1); + expect(storage.abortIncompleteUploads).toHaveBeenCalledWith(f.companyId, cut, expect.anything()); + expect(incomplete.size).toBe(0); + await uploadWorkspaceSnapshot(f.upload); + expect(storage.abortIncompleteUploads).toHaveBeenCalledTimes(1); + }); + it("adopts the published save when another uploader aborts this attempt's upload", async () => { + const f = await fixture(), put = storage.putFile.bind(storage); + let release!: (error: Error) => void, started!: () => void; + const gate = new Promise((_resolve, reject) => { release = reject; }); + const blocked = new Promise(resolve => { started = resolve; }); + let first = true, aborted: string | null = null; + storage.putFile = async input => { + if (!first) return put(input); + first = false; aborted = input.objectKey!; started(); return gate; + }; + storage.abortIncompleteUploads = vi.fn(async (_companyId: string, key: string) => { + if (key === aborted) release(Object.assign(new Error("NoSuchUpload"), { name: "NoSuchUpload" })); + return 1; + }); + const loser = uploadWorkspaceSnapshot(f.upload); + await blocked; + const winner = await uploadWorkspaceSnapshot(f.upload); + expect(storage.abortIncompleteUploads).toHaveBeenCalledWith(f.companyId, aborted, expect.anything()); + expect(await loser).toEqual(winner); + }); it("retains an unsuccessful operation and rejects a content-changing retry", async () => { const f = await fixture(); storage.putFile = async () => { throw new Error("storage unavailable"); }; await expect(uploadWorkspaceSnapshot(f.upload)).rejects.toThrow("storage unavailable"); diff --git a/server/src/services/remote-workspace-snapshot-storage.ts b/server/src/services/remote-workspace-snapshot-storage.ts index 9bdaf0d6f4f..e20ce211512 100644 --- a/server/src/services/remote-workspace-snapshot-storage.ts +++ b/server/src/services/remote-workspace-snapshot-storage.ts @@ -14,7 +14,7 @@ import { executionWorkspaceSnapshots, type Db } from "@paperclipai/db"; import { localEncryptedProvider } from "../secrets/local-encrypted-provider.js"; import type { StoredSecretVersionMaterial } from "../secrets/types.js"; import type { StorageService } from "../storage/types.js"; -import { notifyRemoteWorkspaceSnapshots, REMOTE_WORKSPACE_SNAPSHOT_GRACE_MS } from "./remote-workspace-snapshot-retention.js"; +import { abortSnapshotAttemptUploads, notifyRemoteWorkspaceSnapshots, snapshotAttemptKeys, REMOTE_WORKSPACE_SNAPSHOT_GRACE_MS } from "./remote-workspace-snapshot-retention.js"; import { WORKSPACE_ARCHIVE_IO_BUFFER_BYTES } from "./remote-workspace-archive-io.js"; const SCHEMA = "paperclip.workspace-snapshot.v1"; @@ -296,6 +296,7 @@ async function uploadVerifiedWorkspaceSnapshot(input: WorkspaceSnapshotUpload, v const winning = updated ?? (await input.db.select().from(executionWorkspaceSnapshots).where(scope(input)))[0]; if (!winning || !["uploaded", "committed", "superseded"].includes(winning.status)) failure("publication_changed"); const saved = reference(winning!); + if (updated) await abortSnapshotAttemptUploads(input.storage, saved.companyId, snapshotAttemptKeys(updated)); if (!recapture && saved.manifestDigest !== input.manifestDigest) failure("operation_content_changed"); if (saved.objectKey !== attempt.objectKey) await verifyTimed(saved); return saved; @@ -376,9 +377,17 @@ async function uploadVerifiedWorkspaceSnapshot(input: WorkspaceSnapshotUpload, v } // The object key is unique per attempt, so a late losing uploader cannot // overwrite an object already used by a committed generation. - const stored = await timedStep(input.timing, "putMs", "storage.snapshot.put", () => input.storage.putFile({ companyId: identity.companyId, namespace: "workspace-snapshots", objectKey: attempt.objectKey, - originalFilename: null, contentType: "application/octet-stream", body: createReadStream(encrypted, { highWaterMark: WORKSPACE_ARCHIVE_IO_BUFFER_BYTES }), - byteSize: attempt.byteSize, sha256: attempt.sha256, sourceFile: encrypted, signal: input.signal })); + let stored: Awaited>; + try { + stored = await timedStep(input.timing, "putMs", "storage.snapshot.put", () => input.storage.putFile({ companyId: identity.companyId, namespace: "workspace-snapshots", objectKey: attempt.objectKey, + originalFilename: null, contentType: "application/octet-stream", body: createReadStream(encrypted, { highWaterMark: WORKSPACE_ARCHIVE_IO_BUFFER_BYTES }), + byteSize: attempt.byteSize, sha256: attempt.sha256, sourceFile: encrypted, signal: input.signal })); + } catch (error) { + if (input.signal?.aborted) throw error; + const saved = await getWorkspaceSnapshot(input.db, input); + if (!saved || (!recapture && saved.manifestDigest !== input.manifestDigest)) throw error; + await verifyTimed(saved); return saved; + } const uploaded = { schema: SCHEMA, ...identity, ...attempt } as WorkspaceSnapshotRef; // The backend checked every part of this upload against digests of these // exact bytes, so only its size needs confirming. Any other provider, and diff --git a/server/src/shutdown.ts b/server/src/shutdown.ts index 475df7c095b..894a05f2e35 100644 --- a/server/src/shutdown.ts +++ b/server/src/shutdown.ts @@ -14,6 +14,11 @@ export type ShutdownStepResult = | { outcome: "timed_out"; durationMs: number } | { outcome: "failed"; error: unknown; durationMs: number }; +export async function afterStalledShutdownStep(result: ShutdownStepResult, onStalled: () => Promise) { + if (result.outcome !== "completed") await onStalled(); + return result; +} + export async function runBoundedShutdownStep(input: { step: string; signal: "SIGINT" | "SIGTERM"; diff --git a/server/src/storage/s3-multipart.ts b/server/src/storage/s3-multipart.ts index 2f64817bc09..515faa56d45 100644 --- a/server/src/storage/s3-multipart.ts +++ b/server/src/storage/s3-multipart.ts @@ -9,11 +9,34 @@ import { PassThrough, Readable } from "node:stream"; import { setTimeout as delay } from "node:timers/promises"; import type { PutObjectInput, PutObjectReceipt } from "./types.js"; import { withS3FailureDiagnostics } from "./s3-failure.js"; -import { cleanupStorageFailure } from "./failure-cleanup.js"; +import { cleanupStorageFailure, STORAGE_FAILURE_CLEANUP_MS } from "./failure-cleanup.js"; const bufferedPartSize = 8 * 1024 * 1024; const bufferedPayloadLimit = 4 * bufferedPartSize; let bufferedPayloadBytes = 0; +const inFlightUploads = new Set<{ abort: (reason: unknown) => void; settled: Promise }>(); + +function trackUpload(abort: (reason: unknown) => void) { + let settle!: () => void; + const entry = { abort, settled: new Promise(resolve => { settle = resolve; }) }; + inFlightUploads.add(entry); + return () => { inFlightUploads.delete(entry); settle(); }; +} + +function settledWithin(entries: Array<{ settled: Promise }>, ms: number) { + let timer: ReturnType | undefined; + return Promise.race([Promise.all(entries.map(entry => entry.settled)).then(() => undefined), + new Promise(resolve => { timer = setTimeout(resolve, ms); })]).finally(() => clearTimeout(timer)); +} + +export async function abortInFlightMultipartUploads(reason: unknown, options: { graceMs?: number; timeoutMs?: number } = {}) { + if (options.graceMs && inFlightUploads.size) await settledWithin([...inFlightUploads], options.graceMs); + const entries = [...inFlightUploads]; + if (!entries.length) return 0; + for (const entry of entries) entry.abort(reason); + await settledWithin(entries, options.timeoutMs ?? 2 * STORAGE_FAILURE_CLEANUP_MS); + return entries.length; +} /** Each part is streamed too: multipart must not become another whole-file * buffer or a fixed part-count ceiling below the storage provider's limits. */ @@ -37,6 +60,7 @@ export async function putS3Multipart(client: S3Client, bucket: string, key: stri const onAbort = () => stop(input.signal!.reason); input.signal?.addEventListener("abort", onAbort, { once: true }); if (input.signal?.aborted) onAbort(); + const untrack = trackUpload(stop); const readNext = async () => { controller.signal.throwIfAborted(); let abort!: () => void; @@ -126,6 +150,7 @@ export async function putS3Multipart(client: S3Client, bucket: string, key: stri input.signal?.removeEventListener("abort", onAbort); source.destroy(); await cleanupStorageFailure(async () => iterator.return?.()); + untrack(); } } @@ -211,6 +236,7 @@ export async function putS3MultipartFile( const onAbort = () => controller.abort(input.signal!.reason); input.signal?.addEventListener("abort", onAbort, { once: true }); if (input.signal?.aborted) onAbort(); + const untrack = trackUpload(reason => { if (!controller.signal.aborted) controller.abort(reason); }); // This path never reads the stream body; release its file descriptor now. if (input.body instanceof Readable) input.body.destroy(); let uploadId: string | undefined; @@ -283,5 +309,6 @@ export async function putS3MultipartFile( throw error; } finally { input.signal?.removeEventListener("abort", onAbort); + untrack(); } } diff --git a/server/src/storage/s3-provider.ts b/server/src/storage/s3-provider.ts index 7ff5012f4b4..3d0cdd8e531 100644 --- a/server/src/storage/s3-provider.ts +++ b/server/src/storage/s3-provider.ts @@ -6,6 +6,8 @@ import { PutObjectCommand, ListObjectsV2Command, ListObjectVersionsCommand, + ListMultipartUploadsCommand, + AbortMultipartUploadCommand, } from "@aws-sdk/client-s3"; import { assertSourceFileSize, fileRangeMd5, putS3Multipart, putS3MultipartFile, sendFileRequestWithRetry } from "./s3-multipart.js"; import { createReadStream } from "node:fs"; @@ -198,6 +200,33 @@ export function createS3StorageProvider(config: S3ProviderConfig): StorageProvid return { etag: output.ETag }; }, + async abortIncompleteUploads(input) { + const key = buildKey(prefix, input.objectKey); + const uploads: string[] = []; + let keyMarker: string | undefined, uploadIdMarker: string | undefined; + for (let pages = 0; ; pages++) { + input.signal?.throwIfAborted(); + if (pages >= 100) throw new Error("s3_multipart_listing_cursor_invalid"); + const output = await client.send(new ListMultipartUploadsCommand({ Bucket: bucket, Prefix: key, + KeyMarker: keyMarker, UploadIdMarker: uploadIdMarker, MaxUploads: 1000 }), { abortSignal: input.signal }); + for (const upload of output.Uploads ?? []) if (upload.Key === key && upload.UploadId) uploads.push(upload.UploadId); + if (!output.IsTruncated) break; + if (!output.NextKeyMarker && !output.NextUploadIdMarker) throw new Error("s3_multipart_listing_cursor_invalid"); + keyMarker = output.NextKeyMarker; uploadIdMarker = output.NextUploadIdMarker; + } + let aborted = 0; + for (const uploadId of uploads) { + input.signal?.throwIfAborted(); + try { + await client.send(new AbortMultipartUploadCommand({ Bucket: bucket, Key: key, UploadId: uploadId }), { abortSignal: input.signal }); + aborted++; + } catch (error) { + if ((error as { name?: string }).name !== "NoSuchUpload") throw error; + } + } + return aborted; + }, + async listObjects(input) { const keyPrefix = buildKey(prefix, input.prefix); const output = await client.send(new ListObjectsV2Command({ diff --git a/server/src/storage/service.ts b/server/src/storage/service.ts index 9203903d0d5..e55afb81b33 100644 --- a/server/src/storage/service.ts +++ b/server/src/storage/service.ts @@ -158,5 +158,11 @@ export function createStorageService(provider: StorageProvider): StorageService await cancellable(options?.signal, () => provider.deleteObject({ objectKey, signal: options?.signal, ...(options?.allVersions ? { allVersions: true } : {}) })); }, + + async abortIncompleteUploads(companyId: string, objectKey: string, options) { + ensureCompanyPrefix(companyId, objectKey); + if (!provider.abortIncompleteUploads) return 0; + return cancellable(options?.signal, () => provider.abortIncompleteUploads!({ objectKey, signal: options?.signal })); + }, }; } diff --git a/server/src/storage/types.ts b/server/src/storage/types.ts index cec0dcdb473..c45f36b560b 100644 --- a/server/src/storage/types.ts +++ b/server/src/storage/types.ts @@ -72,6 +72,8 @@ export interface StorageProvider { | { ifMatch: string; ifNoneMatch?: never } )): Promise<{ etag: string }>; + abortIncompleteUploads?(input: { objectKey: string; signal?: AbortSignal }): Promise; + listObjects?(input: { prefix: string; continuationToken?: string; signal?: AbortSignal }): Promise<{ keys: string[]; continuationToken?: string; @@ -106,4 +108,5 @@ export interface StorageService { getObject(companyId: string, objectKey: string, options?: Pick): Promise; headObject(companyId: string, objectKey: string, options?: Pick): Promise; deleteObject(companyId: string, objectKey: string, options?: Pick): Promise; + abortIncompleteUploads?(companyId: string, objectKey: string, options?: { signal?: AbortSignal }): Promise; } diff --git a/tests/restart-chaos/storage-orphan-uploads.test.ts b/tests/restart-chaos/storage-orphan-uploads.test.ts new file mode 100644 index 00000000000..03428e9a699 --- /dev/null +++ b/tests/restart-chaos/storage-orphan-uploads.test.ts @@ -0,0 +1,54 @@ +import { mkdtemp, rm } from "node:fs/promises"; +import { createRequire } from "node:module"; +import os from "node:os"; +import path from "node:path"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; +import { createS3StorageProvider } from "../../server/src/storage/s3-provider.js"; +import { minioBinary, startMinio, type ObjectStore } from "./object-store.js"; +import { listIncompleteUploads } from "./storage-audit.js"; + +const binary = minioBinary(); +const sdk = createRequire(new URL("../../server/package.json", import.meta.url))("@aws-sdk/client-s3"); +const MIB = 1024 * 1024; + +describe.skipIf(!binary)("incomplete snapshot uploads against MinIO", () => { + let root: string; + let store: ObjectStore; + let client: InstanceType; + const prefix = "chaos/w-orphan"; + const snapshot = "company/workspace-snapshots/ws/snapshot"; + + beforeAll(async () => { + root = await mkdtemp(path.join(os.tmpdir(), "restart-chaos-orphan-")); + store = await startMinio({ binary: binary!, root }); + client = new sdk.S3Client({ endpoint: store.endpoint, region: store.region, forcePathStyle: true, maxAttempts: 1, + credentials: { accessKeyId: store.accessKeyId, secretAccessKey: store.secretAccessKey } }); + }, 60_000); + afterEach(() => { vi.unstubAllEnvs(); }); + afterAll(async () => { + client?.destroy(); + await store?.stop(); + await rm(root, { recursive: true, force: true }); + }); + + async function cutUpload(key: string) { + const { UploadId } = await client.send(new sdk.CreateMultipartUploadCommand({ Bucket: store.bucket, Key: `${prefix}/${key}` })); + await client.send(new sdk.UploadPartCommand({ Bucket: store.bucket, Key: `${prefix}/${key}`, UploadId, PartNumber: 1, Body: Buffer.alloc(5 * MIB, 1) })); + return UploadId as string; + } + + it("aborts only the cut-off attempt's uploads and leaves a sibling attempt and a longer key alone", async () => { + vi.stubEnv("AWS_ACCESS_KEY_ID", store.accessKeyId); + vi.stubEnv("AWS_SECRET_ACCESS_KEY", store.secretAccessKey); + vi.stubEnv("AWS_SESSION_TOKEN", ""); + const cut = `${snapshot}/cut.pcws`, sibling = `${snapshot}/sibling.pcws`, longer = `${snapshot}/cut.pcws.longer`; + await cutUpload(cut); await cutUpload(cut); + const siblingId = await cutUpload(sibling), longerId = await cutUpload(longer); + expect(await listIncompleteUploads(store, `${prefix}/`)).toHaveLength(4); + const provider = createS3StorageProvider({ bucket: store.bucket, region: store.region, endpoint: store.endpoint, prefix, forcePathStyle: true }); + expect(await provider.abortIncompleteUploads!({ objectKey: cut })).toBe(2); + expect((await listIncompleteUploads(store, `${prefix}/`)).map(upload => [upload.key, upload.uploadId]).sort()) + .toEqual([[`${prefix}/${longer}`, longerId], [`${prefix}/${sibling}`, siblingId]].sort()); + expect(await provider.abortIncompleteUploads!({ objectKey: cut })).toBe(0); + }, 60_000); +});