feat(cli): parallel asset publishing (#19367)

Adds parallel asset publishing. In a 60-asset/60-Lambda test project, this change decreases the total hot-swap time from 57 seconds to 18 seconds.

Fixes #19193

----

*By submitting this pull request, I confirm that my contribution is made under the terms of the Apache-2.0 license*

Author: misterjoshua

packages/aws-cdk/lib/util/asset-publishing.ts

@@ -21,6 +21,7 @@ export async function publishAssets(manifest: cdk_assets.AssetManifest, sdk: Sdk
     aws: new PublishingAws(sdk, targetEnv),
     progressListener: new PublishingProgressListener(),
     throwOnError: false,
+    publishInParallel: true,
   });
   await publisher.publish();
   if (publisher.hasFailures) {

packages/cdk-assets/lib/publishing.ts

@@ -23,6 +23,13 @@ export interface AssetPublishingOptions {
    * @default true
    */
   readonly throwOnError?: boolean;
+
+  /**
+   * Whether to publish in parallel
+   *
+   * @default false
+   */
+  readonly publishInParallel?: boolean;
 }
 
 /**
@@ -56,44 +63,33 @@ export class AssetPublishing implements IPublishProgress {
   private readonly totalOperations: number;
   private completedOperations: number = 0;
   private aborted = false;
+  private readonly handlerHost: IHandlerHost;
+  private readonly publishInParallel: boolean;
 
   constructor(private readonly manifest: AssetManifest, private readonly options: AssetPublishingOptions) {
     this.assets = manifest.entries;
     this.totalOperations = this.assets.length;
-  }
+    this.publishInParallel = options.publishInParallel ?? false;
 
-  /**
-   * Publish all assets from the manifest
-   */
-  public async publish(): Promise<void> {
     const self = this;
-
-    const handlerHost: IHandlerHost = {
+    this.handlerHost = {
       aws: this.options.aws,
       get aborted() { return self.aborted; },
       emitMessage(t, m) { self.progressEvent(t, m); },
     };
+  }
 
-    for (const asset of this.assets) {
-      if (this.aborted) { break; }
-      this.currentAsset = asset;
-
-      try {
-        if (this.progressEvent(EventType.START, `Publishing ${asset.id}`)) { break; }
-
-        const handler = makeAssetHandler(this.manifest, asset, handlerHost);
-        await handler.publish();
-
-        if (this.aborted) {
-          throw new Error('Aborted');
+  /**
+   * Publish all assets from the manifest
+   */
+  public async publish(): Promise<void> {
+    if (this.publishInParallel) {
+      await Promise.all(this.assets.map(async (asset) => this.publishAsset(asset)));
+    } else {
+      for (const asset of this.assets) {
+        if (!await this.publishAsset(asset)) {
+          break;
         }
-
-        this.completedOperations++;
-        if (this.progressEvent(EventType.SUCCESS, `Published ${asset.id}`)) { break; }
-      } catch (e) {
-        this.failures.push({ asset, error: e });
-        this.completedOperations++;
-        if (this.progressEvent(EventType.FAIL, e.message)) { break; }
       }
     }
 
@@ -102,6 +98,33 @@ export class AssetPublishing implements IPublishProgress {
     }
   }
 
+  /**
+   * Publish an asset.
+   * @param asset The asset to publish
+   * @returns false when publishing should stop
+   */
+  private async publishAsset(asset: IManifestEntry) {
+    try {
+      if (this.progressEvent(EventType.START, `Publishing ${asset.id}`)) { return false; }
+
+      const handler = makeAssetHandler(this.manifest, asset, this.handlerHost);
+      await handler.publish();
+
+      if (this.aborted) {
+        throw new Error('Aborted');
+      }
+
+      this.completedOperations++;
+      if (this.progressEvent(EventType.SUCCESS, `Published ${asset.id}`)) { return false; }
+    } catch (e) {
+      this.failures.push({ asset, error: e });
+      this.completedOperations++;
+      if (this.progressEvent(EventType.FAIL, e.message)) { return false; }
+    }
+
+    return true;
+  }
+
   public get percentComplete() {
     if (this.totalOperations === 0) { return 100; }
     return Math.floor((this.completedOperations / this.totalOperations) * 100);

packages/cdk-assets/test/progress.test.ts

@@ -59,6 +59,19 @@ test('test listener', async () => {
   expect(allMessages).toContain('theAsset:theDestination2');
 });
 
+test('test publishing in parallel', async () => {
+  const progressListener = new FakeListener();
+
+  const pub = new AssetPublishing(AssetManifest.fromPath('/simple/cdk.out'), { aws, progressListener, publishInParallel: true });
+  await pub.publish();
+
+  const allMessages = progressListener.messages.join('\n');
+
+  // Log mentions asset/destination ids
+  expect(allMessages).toContain('theAsset:theDestination1');
+  expect(allMessages).toContain('theAsset:theDestination2');
+});
+
 test('test abort', async () => {
   const progressListener = new FakeListener(true);