feat: treat streaming edge cases

Author: r4mmer

src/new/connection.ts

@@ -25,6 +25,8 @@ enum StreamRequestEvent {
 
 const STREAM_HISTORY_ACK_EVENT = 'request:history:ack';
 
+type FullnodeCapability = 'history-streaming';
+
 /**
  * Stream abort controller that carries the streamId it is managing.
  */
@@ -60,6 +62,8 @@ class WalletConnection extends BaseConnection {
 
   streamWindowSize: number | undefined;
 
+  capabilities?: FullnodeCapability[];
+
   constructor(options: ConnectionParams & { streamWindowSize?: number }) {
     super(options);
 
@@ -96,6 +100,7 @@ class WalletConnection extends BaseConnection {
     this.websocket.on('is_online', this.onConnectionChange);
     this.websocket.on('wallet', this.handleWalletMessage.bind(this));
     this.websocket.on('stream', this.handleStreamMessage.bind(this));
+    this.websocket.on('capabilities', this.handleCapabilities.bind(this));
 
     this.websocket.on('height_updated', height => {
       this.emit('best-block-update', height);
@@ -109,6 +114,29 @@ class WalletConnection extends BaseConnection {
     this.websocket.setup();
   }
 
+  /**
+   * Handle the capabilities event from the websocket.
+   */
+  handleCapabilities(data: { type: string, capabilities: FullnodeCapability[]}) {
+    this.logger.debug(`Fullnode has capabilities: ${JSON.stringify(data.capabilities)}`);
+    const { capabilities } = data;
+    if (!capabilities) {
+      return;
+    }
+    this.capabilities = capabilities;
+  }
+
+  /**
+   * Check if the connected fullnode has the desired capability.
+   * Will return false if the fullnode has not yet sent the capability list.
+   */
+  hasCapability(flag: FullnodeCapability) {
+    if (!this.capabilities) {
+      return false;
+    }
+    return this.capabilities?.includes(flag) || false;
+  }
+
   startControlHandlers(storage: IStorage) {
     this.removeMetricsHandlers();
     this.addMetricsHandlers(storage);

src/new/wallet.js

@@ -1452,7 +1452,7 @@ class HathorWallet extends EventEmitter {
     if (info.network.indexOf(this.conn.network) >= 0) {
       this.storage.setApiVersion(info);
       await this.storage.saveNativeToken();
-      this.conn.start(); // XXX: maybe await?
+      this.conn.start();
     } else {
       this.setState(HathorWallet.CLOSED);
       throw new Error(`Wrong network. server=${info.network} expected=${this.conn.network}`);

src/sync/stream.ts

@@ -1,4 +1,5 @@
 import { Address as BitcoreAddress, HDPublicKey } from 'bitcore-lib';
+import queueMicrotask from 'queue-microtask';
 import FullNodeConnection from '../new/connection';
 import {
   IStorage,
@@ -424,6 +425,12 @@ export class StreamManager extends AbortController {
       throw new Error('No abort controller on connection');
     }
 
+    // The connection with the fullnode may have been lost during the time waiting to sync
+    if(signal.aborted) {
+      this.abort();
+      throw new Error('The connection already aborted this stream');
+    }
+
     signal.addEventListener(
       'abort',
       () => {
@@ -554,6 +561,16 @@ export class StreamManager extends AbortController {
         await this.storage.addTx(item.vertex);
       }
       this.stats.proc();
+
+      /**
+       * This promise will resolve after the JS task queue current run.
+       * This means that we will free IO tasks like receiving events from the WS
+       * and other async code to run before we continue processing our event queue.
+       * @see https://www.npmjs.com/package/queue-microtask
+       */
+      new Promise<void>(resolve => {
+        queueMicrotask(resolve);
+      });
     }
 
     this.isProcessingQueue = false;