fixed asset download oom issue

.talismanrc

@@ -1,4 +1,6 @@
 fileignoreconfig:
   - filename: pnpm-lock.yaml
     checksum: 07642e8dd04d580185a459e5b088d8a1bb4e91be4e04f4842bf4fe4775205bf6
+  - filename: packages/contentstack-export/test/unit/export/modules/assets.test.ts
+    checksum: fda59c011c8cf262bc66dfee5b4e36b7ddc425fa14087d5fc65b83c0cab99197
 version: '1.0'

packages/contentstack-export/src/export/modules/assets.ts

@@ -10,6 +10,8 @@ import isEmpty from 'lodash/isEmpty';
 import includes from 'lodash/includes';
 import progress from 'progress-stream';
 import { createWriteStream } from 'node:fs';
+import { finished as streamFinished } from 'node:stream';
+import { promisify } from 'node:util';
 import { resolve as pResolve } from 'node:path';
 import {
   FsUtility,
@@ -24,6 +26,8 @@ import config from '../../config';
 import { ModuleClassParams } from '../../types';
 import BaseClass, { CustomPromiseHandler, CustomPromiseHandlerInput } from './base-class';
 
+const finished = promisify(streamFinished);
+
 export default class ExportAssets extends BaseClass {
   private assetsRootPath: string;
   public assetConfig = config.modules.assets;
@@ -332,7 +336,7 @@ export default class ExportAssets extends BaseClass {
     const apiBatches: Array<any> = chunk(listOfAssets, this.assetConfig.downloadLimit);
     const downloadedAssetsDirs = await getDirectories(pResolve(this.assetsRootPath, 'files'));
 
-    const onSuccess = ({ response: { data }, additionalInfo }: any) => {
+    const onSuccess = async ({ response: { data }, additionalInfo }: any) => {
       const { asset } = additionalInfo;
       const assetFolderPath = pResolve(this.assetsRootPath, 'files', asset.uid);
       const assetFilePath = pResolve(assetFolderPath, asset.filename);
@@ -344,21 +348,7 @@ export default class ExportAssets extends BaseClass {
       }
 
       const assetWriterStream = createWriteStream(assetFilePath);
-      assetWriterStream.on('error', (error) => {
-        handleAndLogError(
-          error,
-          { ...this.exportConfig.context, uid: asset.uid, filename: asset.fileName },
-          messageHandler.parse('ASSET_DOWNLOAD_FAILED', asset.filename, asset.uid),
-        );
-      });
-      /**
-       * NOTE if pipe not working as expected add the following code below to fix the issue
-       * https://oramind.com/using-streams-efficiently-in-nodejs/
-       * import * as stream from "stream";
-       * import { promisify } from "util";
-       * const finished = promisify(stream.finished);
-       * await finished(assetWriterStream);
-       */
+
       if (this.assetConfig.enableDownloadStatus) {
         const str = progress({
           time: 5000,
@@ -372,7 +362,22 @@ export default class ExportAssets extends BaseClass {
         data.pipe(assetWriterStream);
       }
 
-      log.success(messageHandler.parse('ASSET_DOWNLOAD_SUCCESS', asset.filename, asset.uid), this.exportConfig.context);
+      try {
+        // Wait for the file to actually finish writing before this download is
+        // considered "done" - otherwise downloadLimit never gates real in-flight
+        // transfers, since the response/write streams pile up unboundedly.
+        await finished(assetWriterStream);
+        log.success(
+          messageHandler.parse('ASSET_DOWNLOAD_SUCCESS', asset.filename, asset.uid),
+          this.exportConfig.context,
+        );
+      } catch (error) {
+        handleAndLogError(
+          error,
+          { ...this.exportConfig.context, uid: asset.uid, filename: asset.filename },
+          messageHandler.parse('ASSET_DOWNLOAD_FAILED', asset.filename, asset.uid),
+        );
+      }
     };
 
     const onReject = ({ error, additionalInfo }: any) => {

packages/contentstack-export/test/unit/export/modules/assets.test.ts

@@ -1,3 +1,7 @@
+import * as os from 'os';
+import * as path from 'path';
+import * as fs from 'fs';
+import { PassThrough } from 'stream';
 import { expect } from 'chai';
 import sinon from 'sinon';
 import { FsUtility, getDirectories } from '@contentstack/cli-utilities';
@@ -566,6 +570,63 @@ describe('ExportAssets', () => {
     });
   });
 
+  describe('downloadAssets() concurrency gating (OOM regression)', () => {
+    let tmpDir: string;
+    let makeConcurrentCallStub: sinon.SinonStub;
+
+    beforeEach(() => {
+      tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), 'asset-download-test-'));
+      (exportAssets as any).assetsRootPath = tmpDir;
+      sinon.stub(require('@contentstack/cli-utilities'), 'getDirectories').resolves([]);
+      sinon.stub(FsUtility.prototype, 'getPlainMeta').returns({
+        'file-1': [{ uid: 'asset-1', url: 'https://cdn.example.com/asset-1', filename: 'test.jpg' }],
+      });
+      // Bypass makeConcurrentCall's batching/throttling (it enforces a >=1s floor
+      // per batch via logMsgAndWaitIfRequired, unrelated to this bug) and instead
+      // drive the promisifyHandler directly, the same way makeConcurrentCall would
+      // for a single-item batch. This isolates the actual fix under test: does the
+      // per-asset promise wait for the write to finish, or resolve as soon as
+      // piping starts.
+      makeConcurrentCallStub = sinon.stub(exportAssets as any, 'makeConcurrentCall');
+      makeConcurrentCallStub.callsFake((_env: any, promisifyHandler: any) =>
+        promisifyHandler({ index: 0, batchIndex: 0, isLastRequest: true }),
+      );
+    });
+
+    afterEach(() => {
+      sinon.restore();
+      fs.rmSync(tmpDir, { recursive: true, force: true });
+    });
+
+    it('does not resolve a download until the file has actually finished writing to disk', async () => {
+      const passthrough = new PassThrough();
+      mockStackClient.asset = sinon.stub().returns({
+        download: sinon.stub().resolves({ data: passthrough }),
+      });
+
+      const downloadPromise = exportAssets.downloadAssets();
+      let settled = false;
+      downloadPromise.then(() => {
+        settled = true;
+      });
+
+      // Let the microtask/event-loop queue drain as far as it can without the
+      // write stream ever finishing.
+      await new Promise((r) => setImmediate(r));
+      await new Promise((r) => setImmediate(r));
+
+      // Before the fix, the download "completed" as soon as piping started
+      // (headers received), so this would already be true here - which is
+      // exactly what let unbounded concurrent downloads pile up and OOM.
+      expect(settled).to.equal(false);
+
+      passthrough.end();
+      await downloadPromise;
+
+      expect(settled).to.equal(true);
+    });
+  });
+
   describe('Edge Cases', () => {
     it('should handle empty assets count', async () => {
       const getAssetsCountStub = sinon.stub(exportAssets, 'getAssetsCount').resolves(0);