feat: implement concurrency limit for chunk uploads and update version to 2.0.11

bin/utils/uploadToIpfsSplit.ts

@@ -26,6 +26,7 @@ const COMPLETE_TO_STATUS_DELAY = 5000;
 const PROGRESS_UPDATE_INTERVAL = 200; // ms
 const EXPECTED_UPLOAD_TIME = 60000; // 60 seconds
 const MAX_PROGRESS = 0.9; // 90%
+const MAX_CONCURRENT_CHUNK_UPLOADS = 5;
 
 // Type definitions
 interface ChunkSessionResponse {
@@ -545,6 +546,25 @@ async function delayWithAbortCheck(
   });
 }
 
+async function runTasksWithConcurrency(
+  tasks: Array<() => Promise<void>>,
+  limit: number,
+): Promise<PromiseSettledResult<void>[]> {
+  let nextIndex = 0;
+
+  async function worker(): Promise<void> {
+    while (nextIndex < tasks.length) {
+      const task = tasks[nextIndex++];
+      await task();
+    }
+  }
+
+  const workerCount = Math.min(limit, tasks.length);
+  return Promise.allSettled(
+    Array.from({ length: workerCount }, () => worker()),
+  );
+}
+
 async function uploadFileChunks(
   filePath: string,
   sessionId: string,
@@ -596,7 +616,10 @@ async function uploadFileChunks(
   });
 
   try {
-    const results = await Promise.allSettled(uploadTasks.map((task) => task()));
+    const results = await runTasksWithConcurrency(
+      uploadTasks,
+      MAX_CONCURRENT_CHUNK_UPLOADS,
+    );
     const failedResults = results.filter(
       (result) => result.status === 'rejected',
     );

package-lock.json

@@ -1,12 +1,12 @@
 {
   "name": "pinme",
-  "version": "2.0.10",
+  "version": "2.0.11",
   "lockfileVersion": 3,
   "requires": true,
   "packages": {
     "": {
       "name": "pinme",
-      "version": "2.0.10",
+      "version": "2.0.11",
       "license": "MIT",
       "dependencies": {
         "adm-zip": "^0.5.17",

package.json

@@ -1,6 +1,6 @@
 {
   "name": "pinme",
-  "version": "2.0.10",
+  "version": "2.0.11",
   "publishConfig": {
     "access": "public"
   },

test/cli/success.test.ts

@@ -1,5 +1,6 @@
 import path from 'path';
 import { chmod, mkdir, readFile, writeFile } from 'fs/promises';
+import { setTimeout as delay } from 'timers/promises';
 import AdmZip from 'adm-zip';
 import { describe, expect, test } from 'vitest';
 import {
@@ -209,6 +210,114 @@ describe('pinme CLI success paths with local APIs', () => {
     }
   });
 
+  test('upload limits concurrent chunk uploads to five', async () => {
+    const temp = await createTempHome();
+    let bundle: Awaited<ReturnType<typeof buildCliWithEnv>> | undefined;
+    let activeUploads = 0;
+    let maxActiveUploads = 0;
+    const server = await startLocalHttpServer(async (request, response) => {
+      const bodyText = request.body.toString('utf8');
+
+      if (request.method === 'POST' && request.url === '/chunk/init') {
+        response.writeHead(200, { 'Content-Type': 'application/json' });
+        response.end(
+          JSON.stringify({
+            code: 200,
+            data: {
+              session_id: 'limited-session-1',
+              total_chunks: 8,
+              chunk_size: 1,
+            },
+          }),
+        );
+        return;
+      }
+
+      if (request.method === 'POST' && request.url === '/chunk/upload') {
+        expect(bodyText).toContain('limited-session-1');
+        activeUploads++;
+        maxActiveUploads = Math.max(maxActiveUploads, activeUploads);
+        await delay(50);
+        activeUploads--;
+
+        response.writeHead(200, { 'Content-Type': 'application/json' });
+        response.end(
+          JSON.stringify({
+            code: 200,
+            data: { chunk_index: 0, chunk_size: request.body.length },
+          }),
+        );
+        return;
+      }
+
+      if (request.method === 'POST' && request.url === '/chunk/complete') {
+        response.writeHead(200, { 'Content-Type': 'application/json' });
+        response.end(
+          JSON.stringify({
+            code: 200,
+            data: { trace_id: 'limited-trace-1' },
+          }),
+        );
+        return;
+      }
+
+      if (
+        request.method === 'GET' &&
+        request.url ===
+          '/up_status?trace_id=limited-trace-1&uid=0x1234567890abcdef'
+      ) {
+        response.writeHead(200, { 'Content-Type': 'application/json' });
+        response.end(
+          JSON.stringify({
+            code: 200,
+            data: {
+              is_ready: true,
+              upload_rst: {
+                Bytes: 8,
+                Name: 'multi-chunk.txt',
+                Size: 8,
+                Hash: 'bafy-limited-success',
+              },
+            },
+          }),
+        );
+        return;
+      }
+
+      response.writeHead(404, { 'Content-Type': 'application/json' });
+      response.end(JSON.stringify({ code: 404, msg: 'unexpected request' }));
+    });
+
+    try {
+      const filePath = path.join(temp.home, 'multi-chunk.txt');
+      await writeFile(filePath, '12345678');
+      bundle = await buildCliWithEnv({
+        IPFS_API_URL: server.baseUrl,
+        PINME_API_BASE: server.baseUrl,
+        POLL_INTERVAL_SECONDS: '0',
+        MAX_POLL_TIME_MINUTES: '1',
+      });
+      await writeAuthConfig(temp.home);
+      const result = await runCli(['upload', filePath], {
+        home: temp.home,
+        cliPath: bundle.cliPath,
+        timeout: 20000,
+      });
+
+      expect(result.exitCode, outputOf(result)).toBe(0);
+      expect(maxActiveUploads).toBeLessThanOrEqual(5);
+      expect(
+        server.requests.filter((request) => request.url === '/chunk/upload'),
+      ).toHaveLength(8);
+    } finally {
+      if (bundle) {
+        await bundle.cleanup();
+      }
+      await server.close();
+      await temp.cleanup();
+    }
+  });
+
   test('bind uploads content and binds a PinMe subdomain', async () => {
     const temp = await createTempHome();
     let bundle: Awaited<ReturnType<typeof buildCliWithEnv>> | undefined;