fix: prevent concurrent publish from producing empty files (#428)

## Summary

Fixes a bug where two concurrent `mops publish` sessions for the same
`package@version` could produce empty files in storage. This happened
with `core@2.3.0` where 3 files (`src/Queue.mo`, `src/Result.mo`,
`src/Runtime.mo`) were stored with empty chunk data.

**Root cause**: the storage canister's `startUpload` overwrites active
uploads via `put()`, so a second session's `startUpload` resets chunks
the first session had already filled. `finishUploads` then committed the
empty chunks.

**Fix (defense in depth):**
- **Storage canister**: validate all chunks are non-empty in
`finishUploads` before committing to permanent storage
- **CLI `parallel.ts`**: propagate errors from concurrent tasks instead
of silently swallowing rejections (and keep `busyThreads` balanced on
failure)

## Test plan

- [x] New test file `storage-publish-bug.test.mo` covers:
- Idempotent `startUpload` reset + `finishUploads` catching empty chunks
  - Successful retry after reset with fresh data
  - Partial chunk upload rejection
  - Normal multi-chunk upload regression
  - `chunkCount=0` (legitimately empty files) regression
- [x] Updated existing `storage.test.mo` and `storage-actor.test.mo` to
upload a chunk before `finishUploads` (required by new validation)
- [x] All 17 test files pass locally; `npm run lint` + `npm run
format:check` pass

Author: Kamirus

backend/storage/storage-canister.mo

@@ -94,8 +94,12 @@ shared ({ caller = parent }) persistent actor class Storage() {
       if (Option.isNull(activeUploadsMeta.get(fileId))) {
         return #err("File '" # fileId # "' is not uploading");
       };
-      if (Option.isNull(activeUploadsChunks.get(fileId))) {
-        return #err("File '" # fileId # "' is not uploading");
+      let ?chunks = activeUploadsChunks.get(fileId) else return #err("File '" # fileId # "' is not uploading");
+
+      for (i in chunks.keys()) {
+        if (chunks[i].size() == 0) {
+          return #err("File '" # fileId # "' has empty chunk at index " # Nat.toText(i));
+        };
       };
     };
 

cli/CHANGELOG.md

@@ -3,6 +3,7 @@
 ## Next
 - Support `MOPS_REGISTRY_HOST` and `MOPS_REGISTRY_CANISTER_ID` environment variables for custom registry endpoints
 - Fix `mops build` crashing with `__wbindgen_malloc` error in bundled CLI distribution
+- Fix `parallel()` swallowing errors from concurrent tasks (e.g. `mops publish` uploads), which could hang or leave failures unreported
 
 ## 2.4.0
 - Support `[build].outputDir` config in `mops.toml` for custom build output directory

cli/parallel.ts

@@ -3,11 +3,15 @@ export async function parallel<T>(
   items: T[],
   fn: (item: T) => Promise<void>,
 ) {
-  return new Promise<void>((resolve) => {
+  return new Promise<void>((resolve, reject) => {
     let busyThreads = 0;
+    let failed = false;
     items = items.slice();
 
     let loop = () => {
+      if (failed) {
+        return;
+      }
       if (!items.length) {
         if (busyThreads === 0) {
           resolve();
@@ -18,10 +22,17 @@ export async function parallel<T>(
         return;
       }
       busyThreads++;
-      fn(items.shift() as T).then(() => {
-        busyThreads--;
-        loop();
-      });
+      fn(items.shift() as T).then(
+        () => {
+          busyThreads--;
+          loop();
+        },
+        (err) => {
+          busyThreads--;
+          failed = true;
+          reject(err);
+        },
+      );
       loop();
     };
     loop();

cli/tests/build.test.ts

@@ -1,4 +1,4 @@
-import { describe, expect, test } from "@jest/globals";
+import { describe, expect, jest, test } from "@jest/globals";
 import { execa } from "execa";
 import { existsSync, rmSync } from "node:fs";
 import path from "path";
@@ -14,6 +14,9 @@ function cleanFixture(cwd: string, ...extras: string[]) {
 }
 
 describe("build", () => {
+  // Several dfx/pocket-ic builds per test; slow CI (e.g. node 20 matrix) can exceed 60s default.
+  jest.setTimeout(120_000);
+
   test("ok", async () => {
     const cwd = path.join(import.meta.dirname, "build/success");
     try {

test/storage-actor.test.mo

@@ -41,6 +41,13 @@ actor {
           },
         );
 
+        await test(
+          "upload chunk",
+          func() : async () {
+            assert Result.isOk(await storage.uploadChunk(fileId, 0, Blob.fromArray([1, 2, 3])));
+          },
+        );
+
         await test(
           "try to finish upload with unknown file id",
           func() : async () {

test/storage-publish-bug.test.mo

@@ -0,0 +1,189 @@
+import Result "mo:base/Result";
+import Blob "mo:base/Blob";
+import { test; suite } "mo:test/async";
+
+import Storage "../backend/storage/storage-canister";
+
+var storage = await Storage.Storage();
+
+// PRECONDITION: startUpload is idempotent — resets active upload (prevents stale data)
+await suite(
+  "PRECONDITION: startUpload reset + finishUploads rejects empty chunks",
+  func() : async () {
+    let fileId = "core@2.3.0/src/Runtime.mo";
+    let realData = Blob.fromArray([1, 2, 3, 4, 5, 6, 7, 8]);
+
+    await test(
+      "start upload and fill chunk",
+      func() : async () {
+        assert Result.isOk(await storage.startUpload({ id = fileId; path = "src/Runtime.mo"; chunkCount = 1; owners = [] }));
+        assert Result.isOk(await storage.uploadChunk(fileId, 0, realData));
+      },
+    );
+
+    await test(
+      "second startUpload resets the active upload (idempotent)",
+      func() : async () {
+        assert Result.isOk(await storage.startUpload({ id = fileId; path = "src/Runtime.mo"; chunkCount = 1; owners = [] }));
+      },
+    );
+
+    await test(
+      "finishUploads rejects because reset cleared the chunk data",
+      func() : async () {
+        assert Result.isErr(await storage.finishUploads([fileId]));
+      },
+    );
+  },
+);
+
+// PRECONDITION: idempotent startUpload allows a clean retry to succeed
+await suite(
+  "PRECONDITION: retry after reset succeeds with fresh data",
+  func() : async () {
+    storage := await Storage.Storage();
+    let fileId = "core@2.3.0/src/Runtime.mo";
+    let staleData = Blob.fromArray([1, 2, 3]);
+    let freshData = Blob.fromArray([10, 20, 30, 40, 50]);
+
+    await test(
+      "stale session: start and upload",
+      func() : async () {
+        assert Result.isOk(await storage.startUpload({ id = fileId; path = "src/Runtime.mo"; chunkCount = 1; owners = [] }));
+        assert Result.isOk(await storage.uploadChunk(fileId, 0, staleData));
+      },
+    );
+
+    await test(
+      "retry session: startUpload resets, then upload fresh data",
+      func() : async () {
+        assert Result.isOk(await storage.startUpload({ id = fileId; path = "src/Runtime.mo"; chunkCount = 1; owners = [] }));
+        assert Result.isOk(await storage.uploadChunk(fileId, 0, freshData));
+      },
+    );
+
+    await test(
+      "finishUploads succeeds with fresh data",
+      func() : async () {
+        assert Result.isOk(await storage.finishUploads([fileId]));
+      },
+    );
+
+    await test(
+      "downloaded file has the fresh data",
+      func() : async () {
+        let chunkRes = await storage.downloadChunk(fileId, 0);
+        switch (chunkRes) {
+          case (#ok(chunk)) {
+            assert chunk == freshData;
+          };
+          case (#err(_)) {
+            assert false;
+          };
+        };
+      },
+    );
+  },
+);
+
+// FIX VERIFICATION: finishUploads rejects files with empty chunks
+await suite(
+  "FIX: finishUploads rejects empty chunks",
+  func() : async () {
+    storage := await Storage.Storage();
+    let fileId = "pkg@1.0.0/src/Lib.mo";
+
+    await test(
+      "start upload with chunkCount=2 but upload only chunk 0",
+      func() : async () {
+        assert Result.isOk(await storage.startUpload({ id = fileId; path = "src/Lib.mo"; chunkCount = 2; owners = [] }));
+        assert Result.isOk(await storage.uploadChunk(fileId, 0, Blob.fromArray([10, 20, 30])));
+      },
+    );
+
+    await test(
+      "finishUploads is rejected because chunk 1 was never uploaded",
+      func() : async () {
+        let res = await storage.finishUploads([fileId]);
+        assert Result.isErr(res);
+      },
+    );
+  },
+);
+
+// REGRESSION: normal upload flow still works
+await suite(
+  "REGRESSION: normal upload flow",
+  func() : async () {
+    storage := await Storage.Storage();
+    let fileId = "pkg@2.0.0/src/Main.mo";
+    let data1 = Blob.fromArray([10, 20, 30, 40, 50]);
+    let data2 = Blob.fromArray([60, 70, 80]);
+
+    await test(
+      "upload file with 2 chunks",
+      func() : async () {
+        assert Result.isOk(await storage.startUpload({ id = fileId; path = "src/Main.mo"; chunkCount = 2; owners = [] }));
+        assert Result.isOk(await storage.uploadChunk(fileId, 0, data1));
+        assert Result.isOk(await storage.uploadChunk(fileId, 1, data2));
+      },
+    );
+
+    await test(
+      "finishUploads succeeds when all chunks are present",
+      func() : async () {
+        assert Result.isOk(await storage.finishUploads([fileId]));
+      },
+    );
+
+    await test(
+      "downloaded chunks have correct data",
+      func() : async () {
+        let c0 = await storage.downloadChunk(fileId, 0);
+        switch (c0) {
+          case (#ok(chunk)) { assert chunk == data1 };
+          case (#err(_)) { assert false };
+        };
+        let c1 = await storage.downloadChunk(fileId, 1);
+        switch (c1) {
+          case (#ok(chunk)) { assert chunk == data2 };
+          case (#err(_)) { assert false };
+        };
+      },
+    );
+  },
+);
+
+// REGRESSION: empty files (chunkCount=0) still work
+await suite(
+  "REGRESSION: empty files (chunkCount=0) are allowed",
+  func() : async () {
+    storage := await Storage.Storage();
+    let fileId = "pkg@3.0.0/src/Empty.mo";
+
+    await test(
+      "start upload with chunkCount=0",
+      func() : async () {
+        assert Result.isOk(await storage.startUpload({ id = fileId; path = "src/Empty.mo"; chunkCount = 0; owners = [] }));
+      },
+    );
+
+    await test(
+      "finishUploads succeeds for empty files",
+      func() : async () {
+        assert Result.isOk(await storage.finishUploads([fileId]));
+      },
+    );
+
+    await test(
+      "file meta reports 0 chunks",
+      func() : async () {
+        let res = await storage.getFileMeta(fileId);
+        switch (res) {
+          case (#ok(meta)) { assert meta.chunkCount == 0 };
+          case (#err(_)) { assert false };
+        };
+      },
+    );
+  },
+);

test/storage.test.mo

@@ -33,6 +33,13 @@ await suite(
       },
     );
 
+    await test(
+      "upload chunk",
+      func() : async () {
+        assert Result.isOk(await storage.uploadChunk(fileId, 0, Blob.fromArray([1, 2, 3])));
+      },
+    );
+
     await test(
       "try to finish upload with unknown file id",
       func() : async () {