fix unhandled promise reject during fetch process (teambit#3968)

Fetch process sends the objects to a queue to be written to the filesystem and doesn't wait to completion.
Originally it was done to make the process faster and let the queue control the concurrency. However, it causes issues with error handling.
This PR changes the implementation to wait for the queue to complete processing before continuing to the next object.

The performance had been tested before and after this change and there was no noticeable difference. (on bit-bin with more than 10K objects it took 1:10 min in both cases, which is mostly the download time)

Author: davidfirst

e2e/functionalities/merge.e2e.3.ts

@@ -52,7 +52,8 @@ describe('merge functionality', function () {
     it('should throw MergeConflict error when importing the component', () => {
       const importFunc = () => helper.command.importComponent('bar/foo');
       const error = new MergeConflict(`${helper.scopes.remote}/bar/foo`, ['0.0.2']);
-      helper.general.expectToThrow(importFunc, error);
+      expect(importFunc).to.throw(error.message);
+      expect(importFunc).to.not.throw('unhandled rejection found');
     });
   });
   describe('importing a component with --merge flag', () => {

src/cli/default-error-handler.ts

@@ -80,7 +80,6 @@ import {
   HashNotFound,
   HeadNotFound,
   InvalidIndexJson,
-  MergeConflict,
   OutdatedIndexJson,
   ParentNotFound,
   ResolutionException,
@@ -219,18 +218,6 @@ const errorsMap: Array<[Class<Error>, (err: Class<Error>) => string]> = [
   ],
   [HashNotFound, (err) => `hash ${chalk.bold(err.hash)} not found`],
   [HeadNotFound, (err) => `head snap ${chalk.bold(err.headHash)} was not found for a component ${chalk.bold(err.id)}`],
-  [
-    MergeConflict,
-    (err) =>
-      `error: merge conflict occurred while importing the component ${err.id}. conflict version(s): ${err.versions.join(
-        ', '
-      )}
-to resolve it and merge your local and remote changes, please do the following:
-1) bit untag ${err.id} ${err.versions.join(' ')}
-2) bit import
-3) bit checkout ${err.versions.join(' ')} ${err.id}
-once your changes are merged with the new remote version, you can tag and export a new version of the component to the remote scope.`,
-  ],
   [
     OutdatedIndexJson,
     (err) => `error: ${chalk.bold(err.id)} found in the index.json file, however, is missing from the scope.

src/scope/exceptions/merge-conflict.ts

@@ -1,11 +1,18 @@
-import AbstractError from '../../error/abstract-error';
+import { BitError } from '@teambit/bit-error';
 
-export default class MergeConflict extends AbstractError {
+export default class MergeConflict extends BitError {
   id: string;
   versions: string[];
 
   constructor(id: string, versions: string[]) {
-    super();
+    super(`error: merge conflict occurred while importing the component ${id}. conflict version(s): ${versions.join(
+      ', '
+    )}
+to resolve it and merge your local and remote changes, please do the following:
+1) bit untag ${id} ${versions.join(' ')}
+2) bit import
+3) bit checkout ${versions.join(' ')} ${id}
+once your changes are merged with the new remote version, you can tag and export a new version of the component to the remote scope.`);
     this.id = id;
     this.versions = versions;
   }

src/scope/objects-fetcher/objects-writable-stream.ts

@@ -1,7 +1,6 @@
-import pMap from 'p-map';
 import { Writable } from 'stream';
 import { BitObject, Repository } from '../objects';
-import { CONCURRENT_COMPONENTS_LIMIT, DEFAULT_LANE } from '../../constants';
+import { DEFAULT_LANE } from '../../constants';
 import { RemoteLaneId } from '../../lane-id/lane-id';
 import logger from '../../logger/logger';
 import { ModelComponentMerger } from '../component-ops/model-components-merger';
@@ -21,7 +20,6 @@ import { WriteComponentsQueue } from './write-components-queue';
  * remotes are processed. see @writeManyObjectListToModel.
  */
 export class ObjectsWritable extends Writable {
-  private mutableObjects: BitObject[] = [];
   constructor(
     private repo: Repository,
     private sources: SourceRepository,
@@ -37,46 +35,27 @@ export class ObjectsWritable extends Writable {
       return callback(new Error('objectItem expected to have "ref" and "buffer" props'));
     }
     try {
-      await this.writeImmutableObjectToFs(obj);
+      await this.writeObjectToFs(obj);
       return callback();
     } catch (err) {
       return callback(err);
     }
   }
-  async _final(callback) {
-    try {
-      await this.writeMutableObjectsToFS();
-      callback();
-    } catch (err) {
-      callback(err);
-    }
-  }
-  private async writeImmutableObjectToFs(obj: ObjectItem) {
+
+  private async writeObjectToFs(obj: ObjectItem) {
     const bitObject = await BitObject.parseObject(obj.buffer);
     if (bitObject instanceof Lane) {
       throw new Error('ObjectsWritable does not support lanes');
     }
     if (bitObject instanceof ModelComponent) {
-      this.componentsQueue.addComponent(bitObject.id(), () => this.writeComponentObject(bitObject));
+      await this.componentsQueue.addComponent(bitObject.id(), () => this.writeComponentObject(bitObject));
     } else {
-      this.objectsQueue.addImmutableObject(obj.ref.toString(), () => this.repo.writeObjectsToTheFS([bitObject]));
+      await this.objectsQueue.addImmutableObject(obj.ref.toString(), () => this.writeImmutableObject(bitObject));
     }
-    // else this.mutableObjects.push(bitObject);
   }
-  private async writeMutableObjectsToFS() {
-    const components = this.mutableObjects.filter((obj) => obj instanceof ModelComponent);
-    const mergedComponents = await pMap(
-      components,
-      (component) => this.mergeModelComponent(component as ModelComponent, this.remoteName),
-      {
-        concurrency: CONCURRENT_COMPONENTS_LIMIT,
-      }
-    );
-    await this.repo.writeObjectsToTheFS(mergedComponents);
-    await this.repo.remoteLanes.addEntriesFromModelComponents(
-      RemoteLaneId.from(DEFAULT_LANE, this.remoteName),
-      mergedComponents
-    );
+
+  private async writeImmutableObject(bitObject: BitObject) {
+    await this.repo.writeObjectsToTheFS([bitObject]);
   }
 
   private async writeComponentObject(modelComponent: ModelComponent) {

src/scope/objects-fetcher/write-components-queue.ts

@@ -8,8 +8,7 @@ export class WriteComponentsQueue {
   }
   addComponent(id: string, fn: () => Promise<void>) {
     this.processedIds.push(id);
-    // eslint-disable-next-line @typescript-eslint/no-floating-promises
-    this.add(fn);
+    return this.add(fn);
   }
   getQueue() {
     return this.queue;

src/scope/objects-fetcher/write-objects-queue.ts

@@ -7,13 +7,12 @@ export class WriteObjectsQueue {
   constructor(concurrency = CONCURRENT_IO_LIMIT) {
     this.queue = new PQueue({ concurrency, autoStart: true });
   }
-  addImmutableObject(hash: string, fn: () => Promise<void>) {
+  addImmutableObject<T>(hash: string, fn: () => Promise<T | null>) {
     if (this.addedHashes.includes(hash)) {
-      return;
+      return null;
     }
     this.addedHashes.push(hash);
-    // eslint-disable-next-line @typescript-eslint/no-floating-promises
-    this.add(fn);
+    return this.add(fn);
   }
   getQueue() {
     return this.queue;