Replace p-limit (#4394)

lib/plugins/s3/index.js

@@ -1,14 +1,14 @@
 import path from 'node:path';
 import { promises as fsPromises } from 'node:fs';
 
-import pLimit from 'p-limit';
 import { getLogger } from '@sitespeed.io/log';
 import { S3Client, PutObjectCommand } from '@aws-sdk/client-s3';
 
 import { SitespeedioPlugin } from '@sitespeed.io/plugin';
 import { throwIfMissing } from '../../support/util.js';
 import { recursiveReaddir } from '../../support/fileUtil.js';
 import { getContentType } from './contentType.js';
+import { runWithConcurrencyLimit } from './limit.js';
 
 const log = getLogger('sitespeedio.plugin.s3');
 
@@ -86,31 +86,36 @@ export default class S3Plugin extends SitespeedioPlugin {
 
       try {
         const files = await recursiveReaddir(baseDir, true);
-        const limit = pLimit(s3Options.maxAsyncS3 || 20);
-        const uploadPromises = files.map(file =>
-          limit(() =>
-            uploadFile(
-              file,
-              this.s3Client,
-              s3Options,
-              this.storageManager.getStoragePrefix(),
-              baseDir
-            )
-          )
-        );
-        await Promise.all(uploadPromises);
+        const tasks = files.map(file => async () => {
+          return uploadFile(
+            file,
+            this.s3Client,
+            s3Options,
+            this.storageManager.getStoragePrefix(),
+            baseDir
+          );
+        });
+
+        await runWithConcurrencyLimit(tasks, s3Options.maxAsyncS3 || 20);
 
         if (this.options.copyLatestFilesToBase) {
           const rootPath = path.join(baseDir, '..');
           const directoriesAsArray = rootPath.split(path.sep);
           const rootName = directoriesAsArray.at(-1);
           const latestFiles = await recursiveReaddir(rootPath, true);
-          const latestUploadPromises = latestFiles.map(file =>
-            limit(() =>
-              uploadFile(file, this.s3Client, s3Options, rootName, rootPath)
-            )
+          const latestTasks = latestFiles.map(file => async () => {
+            return uploadFile(
+              file,
+              this.s3Client,
+              s3Options,
+              rootName,
+              rootPath
+            );
+          });
+          await runWithConcurrencyLimit(
+            latestTasks,
+            s3Options.maxAsyncS3 || 20
           );
-          await Promise.all(latestUploadPromises);
         }
 
         log.info('Finished upload to S3');

lib/plugins/s3/limit.js

@@ -0,0 +1,34 @@
+export async function runWithConcurrencyLimit(tasks, limit) {
+  const running = new Set();
+
+  async function runNext() {
+    if (tasks.length === 0) {
+      return;
+    }
+
+    const task = tasks.shift();
+    const promise = task()
+      .catch(error => {
+        throw error;
+      })
+      .finally(() => {
+        running.delete(promise);
+        void runNext();
+      });
+
+    running.add(promise);
+    if (running.size < limit) {
+      void runNext();
+    }
+  }
+
+  const starters = [];
+  for (let index = 0; index < limit && tasks.length > 0; index++) {
+    starters.push(runNext());
+  }
+
+  await Promise.allSettled(starters);
+  if (running.size > 0) {
+    await Promise.allSettled(Array.from(running));
+  }
+}

package.json

@@ -104,7 +104,6 @@
     "markdown": "0.5.0",
     "node-scp": "0.0.23",
     "ora": "8.0.1",
-    "p-limit": "6.1.0",
     "pug": "3.0.3",
     "simplecrawler": "1.1.9",
     "tape": "5.8.1",