Merge branch 'master' into fix/selected_as_input_ttl

Author: r4mmer

__tests__/stream.test.ts

@@ -57,7 +57,7 @@ const SERVER_MOCK_TYPE = {
   unknownId: 'unknown_id',
 };
 
-function makeServerMock(mockServer, mockType) {
+function makeServerMock(mockServer, mockType, sendCapabilities = true) {
   function streamHistoryForSocket(streamId, socket) {
     // Begin event marks the start of a stream
     socket.send(
@@ -100,6 +100,9 @@ function makeServerMock(mockServer, mockType) {
   }
 
   mockServer.on('connection', socket => {
+    if (sendCapabilities) {
+      socket.send(JSON.stringify({ type: 'capabilities', capabilities: ['history-streaming'] }));
+    }
     socket.on('message', data => {
       const jsonData = JSON.parse(data);
       if (jsonData.type === 'subscribe_address') {
@@ -277,4 +280,38 @@ describe('Websocket stream history sync', () => {
       mockServer.stop();
     }
   }, 10000);
+
+  it('should default to POLLING_HTTP_API without capabilities', async () => {
+    const mockServer = new Server('ws://localhost:8080/v1a/ws/');
+    makeServerMock(mockServer, SERVER_MOCK_TYPE.simple, false);
+    const wallet = await startWalletFor(HistorySyncMode.MANUAL_STREAM_WS);
+    wallet.conn.on('stream', data => {
+      // Any stream event should fail the test
+      throw new Error(`Received a stream event: ${JSON.stringify(data)}`);
+    });
+    wallet.on('state', state => {
+      // If the sync fails, fail the test
+      if (state === HathorWallet.ERROR) {
+        throw new Error('Wallet reached an error state');
+      }
+    });
+    try {
+      while (true) {
+        if (wallet.isReady()) {
+          break;
+        }
+        await new Promise(resolve => {
+          setTimeout(resolve, 100);
+        });
+      }
+    } finally {
+      // Stop wallet
+      await wallet.stop({ cleanStorage: true, cleanAddresses: true });
+      mockServer.stop();
+    }
+
+    await expect(wallet.getAddressAtIndex(0)).resolves.toEqual(
+      'WewDeXWyvHP7jJTs7tjLoQfoB72LLxJQqN'
+    );
+  }, 10000);
 });

package-lock.json

@@ -24,6 +24,7 @@
     "level": "8.0.1",
     "lodash": "4.17.21",
     "long": "5.2.3",
+    "queue-microtask": "1.2.3",
     "ws": "8.17.1"
   },
   "scripts": {

package.json

@@ -14,6 +14,7 @@ import { handleSubscribeAddress, handleWsDashboard } from '../utils/connection';
 import { IStorage, ILogger, getDefaultLogger } from '../types';
 
 const STREAM_ABORT_TIMEOUT = 10000; // 10s
+const CAPABILITIES_WAIT_TIMEOUT = 2000; // 2s
 
 /**
  * Event names for requesting stream from fullnode
@@ -25,6 +26,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 +63,8 @@ class WalletConnection extends BaseConnection {
 
   streamWindowSize: number | undefined;
 
+  capabilities?: FullnodeCapability[];
+
   constructor(options: ConnectionParams & { streamWindowSize?: number }) {
     super(options);
 
@@ -96,6 +101,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 +115,42 @@ 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;
+  }
+
+  /**
+   * If the fullnode has not sent the capabilities yet wait a while.
+   */
+  async waitCapabilities() {
+    if (this.capabilities === undefined) {
+      // Wait 2s so the fullnode has some time to send the capabilities envent
+      await new Promise<void>(resolve => {
+        setTimeout(resolve, CAPABILITIES_WAIT_TIMEOUT);
+      });
+    }
+  }
+
+  /**
+   * Check if the connected fullnode has the desired capability.
+   * Will return false if the fullnode has not yet sent the capability list.
+   */
+  async hasCapability(flag: FullnodeCapability) {
+    await this.waitCapabilities();
+    if (!this.capabilities) {
+      return false;
+    }
+    return this.capabilities?.includes(flag) || false;
+  }
+
   startControlHandlers(storage: IStorage) {
     this.removeMetricsHandlers();
     this.addMetricsHandlers(storage);

src/new/connection.ts

@@ -172,6 +172,9 @@ class HathorWallet extends EventEmitter {
       this.storage = new Storage(store);
     }
     this.storage.setLogger(this.logger);
+    /**
+     * @type {import('./connection').default}
+     */
     this.conn = connection;
     this.conn.startControlHandlers(this.storage);
 
@@ -1452,7 +1455,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}`);
@@ -2858,7 +2861,22 @@ class HathorWallet extends EventEmitter {
     if (!(await getSupportedSyncMode(this.storage)).includes(this.historySyncMode)) {
       throw new Error('Trying to use an unsupported sync method for this wallet.');
     }
-    const syncMethod = getHistorySyncMethod(this.historySyncMode);
+    let syncMode = this.historySyncMode;
+    if (
+      [HistorySyncMode.MANUAL_STREAM_WS, HistorySyncMode.XPUB_STREAM_WS].includes(
+        this.historySyncMode
+      ) &&
+      !(await this.conn.hasCapability('history-streaming'))
+    ) {
+      // History sync mode is streaming but fullnode is not streaming capable.
+      // We revert to the http polling default.
+      this.logger.debug(
+        'Either fullnode does not support history-streaming or has not sent a capabilities event'
+      );
+      this.logger.debug('Falling back to http polling API');
+      syncMode = HistorySyncMode.POLLING_HTTP_API;
+    }
+    const syncMethod = getHistorySyncMethod(syncMode);
     // This will add the task to the GLL queue and return a promise that
     // resolves when the task finishes executing
     await GLL.add(async () => {

src/new/wallet.js

@@ -1,4 +1,5 @@
 import { Address as BitcoreAddress, HDPublicKey } from 'bitcore-lib';
+import queueMicrotask from 'queue-microtask';
 import FullNodeConnection from '../new/connection';
 import {
   IStorage,
@@ -556,6 +557,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
+       */
+      await new Promise<void>(resolve => {
+        queueMicrotask(resolve);
+      });
     }
 
     this.isProcessingQueue = false;