use tmp file when too many files

console-api/src/packager/packager.service.ts

@@ -17,6 +17,7 @@ import { createHash } from 'crypto';
 import PQueue from 'p-queue';
 import { LockService } from 'src/lock/lock.service';
 import { InjectRedis, Redis } from '@nestjs-modules/ioredis';
+import { Readable } from 'stream';
 
 interface FileWithHash {
   file: readdirp.EntryInfo;
@@ -239,6 +240,9 @@ export class PackagerService extends ConsoleLogger {
         return archive;
       }
     }
+
+    this.log(`Will archive ${archiveName}.`);
+
     return this.archiveQueue.add(async () => {
       const files = archiveTask.filePaths;
       this.log(`Packaging archive ${archiveName} with ${archiveTask.exactFilePaths.length} files.`);
@@ -262,20 +266,42 @@ export class PackagerService extends ConsoleLogger {
       child.stdout.on('data', (chunk) => {
         hashObject.update(chunk);
       });
-      /* if (existing) {
-          await childPromise;
-          archive.hash = hashObject.digest('hex');
-          return archive;
-        }*/
-      const uploadPromise = this.s3.uploadStream(archiveName, child.stdout, {
-        ContentType: 'application/tar+zstd',
-      });
-      const [, { object }] = await Promise.all([childPromise, uploadPromise]);
-      archive.hash = hashObject.digest('hex');
-      await this.redis.set(`hash:${archive.path}`, archive.hash, 'EX', 60 * 60 * 24);
-      archive.size = object.Size;
 
-      return archive;
+      let uploadStream = child.stdout;
+      let tmpFilename: string;
+
+      try {
+        if (files.length > 5000) {
+          // minio would skew the stream if it's too slow
+
+          // use a tmp file to put the stream
+          tmpFilename = path.join(this.packagerWorkingDirectory, `tmp-${archiveName}.tar.zst`);
+          // put the stream to the tmp file
+          const tmpStream = fs.createWriteStream(tmpFilename);
+          uploadStream.pipe(tmpStream);
+          // wait for the stream to finish
+          await new Promise((resolve, reject) => {
+            tmpStream.on('finish', resolve);
+            tmpStream.on('error', reject);
+          });
+          // open the tmp file as a new stream
+          uploadStream = fs.createReadStream(tmpFilename);
+        }
+
+        const uploadPromise = this.s3.uploadStream(archiveName, uploadStream, {
+          ContentType: 'application/tar+zstd',
+        });
+        const [, { object }] = await Promise.all([childPromise, uploadPromise]);
+        archive.hash = hashObject.digest('hex');
+        await this.redis.set(`hash:${archive.path}`, archive.hash, 'EX', 60 * 60 * 24);
+        archive.size = object.Size;
+
+        return archive;
+      } finally {
+        if (tmpFilename) {
+          await fs.promises.rm(tmpFilename);
+        }
+      }
     });
   }