Merge pull request #6427 from nocodb/fix/job-async

fix: handle async error for streamModelDataAsCsv

packages/nocodb/src/modules/jobs/jobs/export-import/duplicate.processor.ts

@@ -242,7 +242,11 @@ export class DuplicateProcessor {
 
     let handledLinks = [];
 
+    let error = null;
+
     for (const sourceModel of sourceModels) {
+      if (error) break;
+
       const dataStream = new Readable({
         read() {},
       });
@@ -251,13 +255,20 @@ export class DuplicateProcessor {
         read() {},
       });
 
-      this.exportService.streamModelDataAsCsv({
-        dataStream,
-        linkStream,
-        projectId: sourceProject.id,
-        modelId: sourceModel.id,
-        handledMmList: handledLinks,
-      });
+      this.exportService
+        .streamModelDataAsCsv({
+          dataStream,
+          linkStream,
+          projectId: sourceProject.id,
+          modelId: sourceModel.id,
+          handledMmList: handledLinks,
+        })
+        .catch((e) => {
+          this.logger.error(e);
+          dataStream.push(null);
+          linkStream.push(null);
+          error = e;
+        });
 
       const model = await Model.get(findWithIdentifier(idMap, sourceModel.id));
 
@@ -284,6 +295,8 @@ export class DuplicateProcessor {
       );
     }
 
+    if (error) throw error;
+
     // update external models (has bt to this model)
     if (externalModels) {
       for (const sourceModel of externalModels) {
@@ -299,14 +312,23 @@ export class DuplicateProcessor {
           read() {},
         });
 
-        this.exportService.streamModelDataAsCsv({
-          dataStream,
-          linkStream,
-          projectId: sourceProject.id,
-          modelId: sourceModel.id,
-          handledMmList: handledLinks,
-          _fieldIds: fields,
-        });
+        let error = null;
+
+        this.exportService
+          .streamModelDataAsCsv({
+            dataStream,
+            linkStream,
+            projectId: sourceProject.id,
+            modelId: sourceModel.id,
+            handledMmList: handledLinks,
+            _fieldIds: fields,
+          })
+          .catch((e) => {
+            this.logger.error(e);
+            dataStream.push(null);
+            linkStream.push(null);
+            error = e;
+          });
 
         const headers: string[] = [];
         let chunk = [];
@@ -401,6 +423,8 @@ export class DuplicateProcessor {
           });
         });
 
+        if (error) throw error;
+
         elapsedTime(
           hrTime,
           `map existing links to ${model.title}`,

packages/nocodb/src/modules/jobs/jobs/export-import/export.service.ts

@@ -722,15 +722,24 @@ export class ExportService {
           dataStream,
         );
 
+        let error = null;
+
         this.streamModelDataAsCsv({
           dataStream,
           linkStream,
           projectId: project.id,
           modelId: model.id,
           handledMmList,
+        }).catch((e) => {
+          this.logger.error(e);
+          dataStream.push(null);
+          linkStream.push(null);
+          error = e;
         });
 
         await Promise.all([uploadPromise, linkPromise]);
+
+        if (error) throw error;
       }
 
       combinedLinkStream.push(null);