explicitly close Stream

apps/app/src/features/openai/server/services/openai.ts

@@ -343,8 +343,20 @@ class OpenaiService implements IOpenaiService {
     });
 
     pagesStream
+      .on('error', () => {
+        batchStrem.end();
+        createVectorStoreFileStream.end();
+      })
       .pipe(batchStrem)
-      .pipe(createVectorStoreFileStream);
+      .on('error', () => {
+        pagesStream.destroy();
+        createVectorStoreFileStream.end();
+      })
+      .pipe(createVectorStoreFileStream)
+      .on('error', () => {
+        pagesStream.destroy();
+        batchStrem.destroy();
+      });
   }
 
   async rebuildVectorStore(page: HydratedDocument<PageDocument>) {

apps/app/src/migrations/20211227060705-revision-path-to-page-id-schema-migration--fixed-7549.js

@@ -1,6 +1,7 @@
 import { Writable } from 'stream';
 
 import mongoose from 'mongoose';
+import { pages } from 'next/dist/build/templates/app-page';
 import streamToPromise from 'stream-to-promise';
 
 import getPageModel from '~/server/models/page';
@@ -57,8 +58,20 @@ module.exports = {
     });
 
     pagesStream
+      .on('error', () => {
+        batchStrem.end();
+        migratePagesStream.end();
+      })
       .pipe(batchStrem)
-      .pipe(migratePagesStream);
+      .on('error', () => {
+        pagesStream.destroy();
+        migratePagesStream.end();
+      })
+      .pipe(migratePagesStream)
+      .on('error', () => {
+        pagesStream.destroy();
+        batchStrem.destroy();
+      });
 
     await streamToPromise(migratePagesStream);
 
@@ -108,8 +121,20 @@ module.exports = {
     });
 
     pagesStream
+      .on('error', () => {
+        batchStrem.end();
+        migratePagesStream.end();
+      })
       .pipe(batchStrem)
-      .pipe(migratePagesStream);
+      .on('error', () => {
+        pagesStream.destroy();
+        migratePagesStream.end();
+      })
+      .pipe(migratePagesStream)
+      .on('error', () => {
+        pagesStream.destroy();
+        batchStrem.destroy();
+      });
 
     await streamToPromise(migratePagesStream);
 

apps/app/src/server/service/file-uploader/local.ts

@@ -163,7 +163,12 @@ module.exports = function(crowi) {
     // mkdir -p
     mkdir.sync(dirpath);
 
-    const stream = fileStream.pipe(fs.createWriteStream(filePath));
+    const writeStream = fs.createWriteStream(filePath);
+
+    const stream = fileStream
+      .on('error', () => { writeStream.end() })
+      .pipe(writeStream)
+      .on('error', () => { fileStream.destroy() });
     return streamToPromise(stream);
   };
 
@@ -177,7 +182,11 @@ module.exports = function(crowi) {
     const fileStream = new Readable();
     fileStream.push(data);
     fileStream.push(null); // EOF
-    const stream = fileStream.pipe(fs.createWriteStream(absFilePath));
+    const writeStream = fs.createWriteStream(absFilePath);
+    const stream = fileStream
+      .on('error', () => { writeStream.end() })
+      .pipe(writeStream)
+      .on('error', () => { fileStream.destroy() });
     return streamToPromise(stream);
   };