Merge pull request #769 from HathorNetwork/chore/sync-release-v1.12.1

chore: sync release v1.12.1

Author: r4mmer

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "@hathor/wallet-lib",
-  "version": "1.12.0",
+  "version": "1.12.1",
   "description": "Library used by Hathor Wallet",
   "main": "lib/index.js",
   "engines": {

package.json

@@ -1,10 +1,30 @@
+import { EventEmitter } from 'events';
 import PQueue from 'queue-promise';
 import { ILogger } from '../types';
 import { GlobalLoadLockTaskError } from '../errors';
 
-const MAX_CONCURRENT_LOAD_TASKS = 5;
+const MAX_CONCURRENT_LOAD_TASKS = 3;
 
 const GLL = new PQueue({ concurrent: MAX_CONCURRENT_LOAD_TASKS });
+const GLLEvents = new EventEmitter();
+
+/**
+ * PQueue will emit a `resolve` event when a task is resolved.
+ * The data will be the taskId since our tasks always resolve with the taskId.
+ * We then emit another event on GLLEvents specially for this taskId.
+ */
+GLL.on('resolve', data => {
+  GLLEvents.emit(data, true);
+});
+
+/**
+ * PQueue will emit a `reject` event when a task is rejected.
+ * The data will be the taskId and the error since our tasks always reject with them.
+ * We then emit another event on GLLEvents specially for this taskId.
+ */
+GLL.on('reject', data => {
+  GLLEvents.emit(data.taskId, false, data.innerError);
+});
 
 /**
  * Add task to the GLL and return a promise that will resolve or reject when the task resolves or rejects.
@@ -16,15 +36,11 @@ export function addTask(task: () => Promise<void>, logger: ILogger) {
   const startTime = Date.now();
   // This promise will resolve or reject when the task does, so the caller can know his task has ended.
   const promise = new Promise<void>((resolve, reject) => {
-    GLL.on('resolve', data => {
-      if (data === taskId) {
+    GLLEvents.once(taskId, (success, error) => {
+      if (success) {
         resolve();
-      }
-    });
-
-    GLL.on('reject', data => {
-      if (data.taskId === taskId) {
-        reject(data.innerError);
+      } else {
+        reject(error);
       }
     });
   });

src/sync/gll.ts

@@ -17,6 +17,7 @@ const QUEUE_GRACEFUL_SHUTDOWN_LIMIT = 10000;
 interface IStreamSyncHistoryBegin {
   type: 'stream:history:begin';
   id: string;
+  seq: number;
 }
 
 interface IStreamSyncHistoryVertex {
@@ -37,6 +38,7 @@ interface IStreamSyncHistoryAddress {
 interface IStreamSyncHistoryEnd {
   type: 'stream:history:end';
   id: string;
+  seq: number;
 }
 
 interface IStreamSyncHistoryError {
@@ -645,9 +647,10 @@ export class StreamManager extends AbortController {
     }
   }
 
-  endStream() {
+  endStream(seq: number) {
     this.logger.debug('Received end-of-stream event.');
     this.hasReceivedEndStream = true;
+    this.connection.sendStreamHistoryAck(this.streamId, seq);
   }
 
   async shutdown() {
@@ -713,7 +716,7 @@ function buildListener(manager: StreamManager, resolve: () => void) {
       manager.generateNextBatch();
     } else if (isStreamSyncHistoryEnd(wsData)) {
       // cleanup and stop the method.
-      manager.endStream();
+      manager.endStream(wsData.seq);
       resolve();
     } else if (isStreamSyncHistoryError(wsData)) {
       // An error happened on the fullnode, we should stop the stream