diff --git a/.talismanrc b/.talismanrc index 71ea6de83..ce3e6d37a 100644 --- a/.talismanrc +++ b/.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' diff --git a/packages/contentstack-export/src/export/modules/assets.ts b/packages/contentstack-export/src/export/modules/assets.ts index 1a95d5f25..965988a0c 100644 --- a/packages/contentstack-export/src/export/modules/assets.ts +++ b/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 = 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) => { diff --git a/packages/contentstack-export/test/unit/export/modules/assets.test.ts b/packages/contentstack-export/test/unit/export/modules/assets.test.ts index 5509682a4..42ce14bdf 100644 --- a/packages/contentstack-export/test/unit/export/modules/assets.test.ts +++ b/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);