fix(packages): reclaim unadopted multipart uploads

convex/httpApiV1.handlers.test.ts

@@ -16956,6 +16956,126 @@ describe("httpApiV1 handlers", () => {
     expect(runAction).not.toHaveBeenCalled();
   });
 
+  it("waits for late multipart stores before cleaning a failed request", async () => {
+    vi.mocked(requirePackagePublishAuth).mockResolvedValue({
+      kind: "user",
+      userId: "users:1",
+      user: { _id: "users:1", handle: "p" },
+    } as never);
+    let finishLateStore!: (id: string) => void;
+    const lateStore = new Promise<string>((resolve) => {
+      finishLateStore = resolve;
+    });
+    const store = vi
+      .fn()
+      .mockRejectedValueOnce(new Error("storage unavailable"))
+      .mockImplementationOnce(() => lateStore);
+    const remove = vi.fn().mockResolvedValue(undefined);
+    const runAction = vi.fn();
+    const form = packagePublishForm(packagePublishMetadata());
+    form.append("files", new File(["first"], "first.txt"));
+    form.append("files", new File(["late"], "late.txt"));
+    const responsePromise = __handlers.publishPackageV1Handler(
+      makeCtx({
+        runMutation: vi.fn().mockResolvedValue(okRate()),
+        runAction,
+        storage: { store, delete: remove },
+      }),
+      new Request("https://example.com/api/v1/packages", { method: "POST", body: form }),
+    );
+    await vi.waitFor(() => expect(store).toHaveBeenCalledTimes(2));
+    expect(remove).not.toHaveBeenCalled();
+    finishLateStore("storage:late");
+    const response = await responsePromise;
+    expect(response.status).toBe(400);
+    expect(await response.text()).toContain("storage unavailable");
+    expect(remove).toHaveBeenCalledExactlyOnceWith("storage:late");
+    expect(runAction).not.toHaveBeenCalled();
+  });
+
+  it.each([false, true])(
+    "cleans partial tarball extraction without deleting reused artifacts (staged: %s)",
+    async (staged) => {
+      vi.mocked(requirePackagePublishAuth).mockResolvedValue({
+        kind: "user",
+        userId: "users:1",
+        user: { _id: "users:1", handle: "p" },
+      } as never);
+      const pack = npmPackFixture({
+        "package/package.json": JSON.stringify({ name: "demo-plugin", version: "1.0.0" }),
+        "package/openclaw.plugin.json": "{}",
+        "package/README.md": "readme",
+      });
+      let finishLateStore!: (id: string) => void;
+      const lateStore = new Promise<string>((resolve) => {
+        finishLateStore = resolve;
+      });
+      const store = vi.fn();
+      if (!staged) store.mockResolvedValueOnce("storage:new-tarball");
+      store
+        .mockRejectedValueOnce(new Error("partial extraction"))
+        .mockImplementationOnce(() => lateStore)
+        .mockResolvedValueOnce("storage:early");
+      const remove = vi.fn().mockResolvedValue(undefined);
+      const runAction = vi.fn();
+      const form = packagePublishForm(packagePublishMetadata());
+      if (staged) {
+        form.set("clawpack", "storage:retained-tarball");
+        form.set("clawpackUploadTicket", "packagePublishUploadTickets:1");
+      } else form.set("clawpack", new File([bytesToArrayBuffer(pack)], "demo.tgz"));
+      const responsePromise = __handlers.publishPackageV1Handler(
+        makeCtx({
+          runMutation: vi.fn().mockResolvedValue(okRate()),
+          runAction,
+          storage: {
+            store,
+            delete: remove,
+            get: vi.fn().mockResolvedValue(new Blob([bytesToArrayBuffer(pack)])),
+          },
+        }),
+        new Request("https://example.com/api/v1/packages", { method: "POST", body: form }),
+      );
+      await vi.waitFor(() => expect(store).toHaveBeenCalledTimes(staged ? 3 : 4));
+      expect(remove).not.toHaveBeenCalled();
+      finishLateStore("storage:late");
+      expect((await responsePromise).status).toBe(400);
+      expect(remove.mock.calls.flat().sort((a, b) => String(a).localeCompare(String(b)))).toEqual(
+        [...(staged ? [] : ["storage:new-tarball"]), "storage:early", "storage:late"].sort((a, b) =>
+          a.localeCompare(b),
+        ),
+      );
+      expect(runAction).not.toHaveBeenCalled();
+    },
+  );
+
+  it("does not infer deletion authority from a failed publication RPC", async () => {
+    vi.mocked(requirePackagePublishAuth).mockResolvedValue({
+      kind: "user",
+      userId: "users:1",
+      user: { _id: "users:1", handle: "p" },
+    } as never);
+    const remove = vi.fn();
+    const runAction = vi.fn().mockRejectedValue(new Error("RPC response lost"));
+    const form = packagePublishForm(packagePublishMetadata());
+    form.append("files", new File(["{}"], "openclaw.plugin.json"));
+    const response = await __handlers.publishPackageV1Handler(
+      makeCtx({
+        runMutation: vi.fn().mockResolvedValue(okRate()),
+        runAction,
+        storage: { store: vi.fn().mockResolvedValue("storage:request-file"), delete: remove },
+      }),
+      new Request("https://example.com/api/v1/packages", { method: "POST", body: form }),
+    );
+    expect(response.status).toBe(400);
+    expect(remove).not.toHaveBeenCalled();
+    expect(runAction).toHaveBeenCalledWith(
+      expect.anything(),
+      expect.objectContaining({
+        requestStorageIds: ["storage:request-file"],
+      }),
+    );
+  });
+
   it("multipart package publish ignores macOS junk files", async () => {
     vi.mocked(getOptionalApiTokenUserId).mockResolvedValue("users:1" as never);
     vi.mocked(requirePackagePublishAuth).mockResolvedValue({
@@ -17383,6 +17503,7 @@ describe("httpApiV1 handlers", () => {
     expect(runAction).toHaveBeenCalledWith(
       expect.anything(),
       expect.objectContaining({
+        requestStorageIds: ["storage:1", "storage:2", "storage:3"],
         payload: expect.objectContaining({
           artifact: expect.objectContaining({
             kind: "npm-pack",

convex/httpApiV1/packagesV1.ts

@@ -1391,12 +1391,34 @@ function bytesToArrayBuffer(bytes: Uint8Array) {
   return copy.buffer;
 }
 
+async function storeRequestPackageBlob(
+  ctx: ActionCtx,
+  requestStorageIds: Set<Id<"_storage">>,
+  blob: Blob,
+) {
+  const storageId = await ctx.storage.store(blob);
+  requestStorageIds.add(storageId);
+  return storageId;
+}
+
+async function settlePackageFileStores(stores: Array<Promise<StoredPackagePublishFile>>) {
+  // A rejected store does not cancel its siblings. Wait before reclaiming their blobs.
+  const results = await Promise.allSettled(stores);
+  return results.map((result) => {
+    if (result.status === "rejected") throw result.reason;
+    return result.value;
+  });
+}
+
 async function storeClawPackFile(
   ctx: ActionCtx,
   entry: { path: string; bytes: Uint8Array },
+  requestStorageIds: Set<Id<"_storage">>,
 ): Promise<StoredPackagePublishFile> {
   const contentType = defaultStoredPackageContentType();
-  const storageId = await ctx.storage.store(
+  const storageId = await storeRequestPackageBlob(
+    ctx,
+    requestStorageIds,
     new Blob([bytesToArrayBuffer(entry.bytes)], { type: contentType }),
   );
   return {
@@ -1417,12 +1439,17 @@ const CLAWPACK_STORE_BATCH_FILES = 16;
 async function storeClawPackFiles(
   ctx: ActionCtx,
   entries: Array<{ path: string; bytes: Uint8Array }>,
+  requestStorageIds: Set<Id<"_storage">>,
 ) {
   const files: StoredPackagePublishFile[] = [];
   let batch: Array<{ path: string; bytes: Uint8Array }> = [];
   let batchBytes = 0;
   const flush = async () => {
-    files.push(...(await Promise.all(batch.map((entry) => storeClawPackFile(ctx, entry)))));
+    files.push(
+      ...(await settlePackageFileStores(
+        batch.map((entry) => storeClawPackFile(ctx, entry, requestStorageIds)),
+      )),
+    );
     batch = [];
     batchBytes = 0;
   };
@@ -1441,13 +1468,16 @@ async function storeClawPackFiles(
 async function storeUploadedPackageFile(
   ctx: ActionCtx,
   entry: File,
+  requestStorageIds: Set<Id<"_storage">>,
 ): Promise<StoredPackagePublishFile> {
   if (entry.size > MAX_PUBLISH_FILE_BYTES) {
     throw new Error(getPublishFileSizeError(entry.name));
   }
   const buffer = new Uint8Array(await entry.arrayBuffer());
   const contentType = normalizeContentType(entry.type) ?? defaultStoredPackageContentType();
-  const storageId = await ctx.storage.store(
+  const storageId = await storeRequestPackageBlob(
+    ctx,
+    requestStorageIds,
     new Blob([bytesToArrayBuffer(buffer)], { type: contentType }),
   );
   return {
@@ -1530,6 +1560,7 @@ async function buildPackagePublishRequestFromClawPack(
   parsed: ParsedPackageClawPack,
   artifactBytes: Uint8Array,
   artifactStorageId: Id<"_storage">,
+  requestStorageIds: Set<Id<"_storage">>,
 ): Promise<ServerPackagePublishRequest> {
   if (parsed.unpackedSize > MAX_PUBLISH_TOTAL_BYTES) {
     throw new Error(getPublishTotalSizeError("package"));
@@ -1546,7 +1577,7 @@ async function buildPackagePublishRequestFromClawPack(
     npmUnpackedSize: parsed.unpackedSize,
     npmFileCount: parsed.fileCount,
   };
-  const files = await storeClawPackFiles(ctx, parsed.entries);
+  const files = await storeClawPackFiles(ctx, parsed.entries, requestStorageIds);
   return { ...metadata, files, artifact };
 }
 
@@ -1597,6 +1628,7 @@ async function parseMultipartPackagePublish(
   ctx: ActionCtx,
   auth: PackagePublishAuth,
   request: Request,
+  requestStorageIds: Set<Id<"_storage">>,
 ): Promise<ServerPackagePublishRequest> {
   const form = await request.formData();
   for (const field of form.keys()) {
@@ -1645,6 +1677,7 @@ async function parseMultipartPackagePublish(
         parsed,
         artifactBytes,
         tarballPart.storageId,
+        requestStorageIds,
       );
     }
 
@@ -1664,7 +1697,9 @@ async function parseMultipartPackagePublish(
     const artifactBytes = new Uint8Array(await tarballEntry.arrayBuffer());
     const parsed = await parseClawPack(artifactBytes);
     assertClawPackPublicationIdentity(metadata, parsed);
-    const artifactStorageId = await ctx.storage.store(
+    const artifactStorageId = await storeRequestPackageBlob(
+      ctx,
+      requestStorageIds,
       new Blob([bytesToArrayBuffer(artifactBytes)], { type: "application/octet-stream" }),
     );
     return await buildPackagePublishRequestFromClawPack(
@@ -1673,6 +1708,7 @@ async function parseMultipartPackagePublish(
       parsed,
       artifactBytes,
       artifactStorageId,
+      requestStorageIds,
     );
   }
 
@@ -1687,8 +1723,8 @@ async function parseMultipartPackagePublish(
   }
 
   const packageFileParts = fileParts.filter((entry) => !isMacJunkPath(entry.name));
-  const files = await Promise.all(
-    packageFileParts.map((entry) => storeUploadedPackageFile(ctx, entry)),
+  const files = await settlePackageFileStores(
+    packageFileParts.map((entry) => storeUploadedPackageFile(ctx, entry, requestStorageIds)),
   );
   if (files.length === 0) throw new Error("files required");
   return { ...metadata, files };
@@ -2669,24 +2705,34 @@ export async function publishPackageV1Handler(ctx: ActionCtx, request: Request)
   const auth = await requirePackagePublishAuthOrResponse(ctx, request, rate.headers);
   if (!auth.ok) return auth.response;
 
+  const requestStorageIds = new Set<Id<"_storage">>();
+  let publicationOwnsCleanup = false;
   try {
     const contentType = request.headers.get("content-type") ?? "";
     if (!contentType.includes("multipart/form-data")) {
       return text("Package publish requires multipart/form-data", 415, rate.headers);
     }
-    const payload = await parseMultipartPackagePublish(ctx, auth.auth, request);
+    const payload = await parseMultipartPackagePublish(ctx, auth.auth, request, requestStorageIds);
+    // After dispatch, only the action can know whether a release adopted these blobs.
+    // An ambiguous RPC failure is not authority for HTTP cleanup.
+    publicationOwnsCleanup = true;
     const result =
       auth.auth.kind === "user"
         ? await runActionRef(ctx, internalRefs.packages.publishPackageForUserInternal, {
             actorUserId: auth.auth.userId,
             payload,
+            requestStorageIds: [...requestStorageIds],
           })
         : await runActionRef(ctx, internalRefs.packages.publishPackageForTrustedPublisherInternal, {
             publishTokenId: auth.auth.publishToken._id,
             payload,
+            requestStorageIds: [...requestStorageIds],
           });
     return json(result, 200, rate.headers);
   } catch (error) {
+    if (!publicationOwnsCleanup) {
+      await Promise.allSettled([...requestStorageIds].map((id) => ctx.storage.delete(id)));
+    }
     return packagePublishErrorToResponse(error, rate.headers);
   }
 }