chore: bound the parallelism (backport #162)

.projenrc.ts

@@ -118,4 +118,10 @@ project.addTask('shrinkwrap', {
   ],
 });
 
+project.addDevDeps('@cdklabs/eslint-plugin');
+project.eslint?.addPlugins('@cdklabs');
+project.eslint?.addRules({
+  '@cdklabs/promiseall-no-unbounded-parallelism': ['error'],
+});
+
 project.synth();

lib/private/archive.ts

@@ -49,6 +49,8 @@ function writeZipFile(directory: string, outputFile: string): Promise<void> {
     // Append files serially to ensure file order
     for (const file of files) {
       const fullPath = path.resolve(directory, file);
+      // There are exactly 2 promises
+      // eslint-disable-next-line @cdklabs/promiseall-no-unbounded-parallelism
       const [data, stat] = await Promise.all([fs.readFile(fullPath), fs.stat(fullPath)]);
       archive.append(data, {
         name: file,

lib/private/p-limit.ts

@@ -0,0 +1,62 @@
+/**
+ * A minimal  of p-limit that does not bring in new dependencies, and is not ESM.
+ */
+
+type PromiseFactory<A> = () => Promise<A>;
+
+export function pLimit(concurrency: number): PLimit {
+  const queue: Array<[PromiseFactory<any>, (x: any) => void, (reason?: any) => void]> = [];
+  let activeCount = 0;
+  let stopped = false;
+
+  function dispatch() {
+    if (activeCount < concurrency && queue.length > 0) {
+      const [fac, resolve, reject] = queue.shift()!;
+      activeCount++;
+      fac().then(
+        (r) => {
+          // Start a new job before reporting back on the previous one
+          resumeNext();
+          resolve(r);
+        },
+        (e) => {
+          // Start a new job before reporting back on the previous one
+          resumeNext();
+          reject(e);
+        }
+      );
+    }
+  }
+
+  function resumeNext() {
+    activeCount--;
+    if (stopped) {
+      for (const [_, __, reject] of queue) {
+        reject(new Error('Task has been cancelled'));
+      }
+      queue.splice(0, queue.length);
+    }
+    dispatch();
+  }
+
+  const ret = <A>(promiseFactory: PromiseFactory<A>) => {
+    return new Promise<A>((resolve, reject) => {
+      queue.push([promiseFactory, resolve, reject]);
+      dispatch();
+    });
+  };
+  Object.defineProperties(ret, {
+    dispose: {
+      value: () => {
+        stopped = true;
+      },
+    },
+  });
+
+  return ret as PLimit;
+}
+
+interface PLimit {
+  dispose(): void;
+  <A>(promiseFactory: PromiseFactory<A>): Promise<A>;
+}

lib/publishing.ts

@@ -3,6 +3,7 @@ import { IAws } from './aws';
 import { IAssetHandler, IHandlerHost, type PublishOptions } from './private/asset-handler';
 import { DockerFactory } from './private/docker';
 import { makeAssetHandler } from './private/handlers';
+import { pLimit } from './private/p-limit';
 import { EventType, IPublishProgress, IPublishProgressListener } from './progress';
 
 export interface AssetPublishingOptions {
@@ -119,7 +120,11 @@ export class AssetPublishing implements IPublishProgress {
    */
   public async publish(options: PublishOptions = {}): Promise<void> {
     if (this.publishInParallel) {
-      await Promise.all(this.assets.map(async (asset) => this.publishAsset(asset, options)));
+      const limit = pLimit(20);
+      // eslint-disable-next-line @cdklabs/promiseall-no-unbounded-parallelism
+      await Promise.all(
+        this.assets.map((asset) => limit(async () => this.publishAsset(asset, options)))
+      );
     } else {
       for (const asset of this.assets) {
         if (!(await this.publishAsset(asset, options))) {