Long poll TSS sessions

packages/bitcore-cli/src/tss.ts

@@ -49,6 +49,15 @@ export async function sign(args: {
     tssKey: walletData.key as TssKeyType
   });
 
+  // Reduces the noise of transient connection errors
+  const connResilience = (count: number, e: Error): number => {
+    if (count > 10) {
+      prompt.log.warn(e.message);
+      return 0;
+    }
+    return ++count;
+  };
+
   // Restore a previously-interrupted TSS session if it exists
   if (fs.existsSync(storedSessionFile)) {
     const storedSession = Encryption.decryptWithPassword(fs.readFileSync(storedSessionFile, 'utf8'), password);
@@ -57,27 +66,36 @@ export async function sign(args: {
 
   // ...otherwise, start a new TSS session
   } else {
-    try {
-      await tssSign.start({
-        id,
-        messageHash,
-        derivationPath,
-        password
-      });
-      storeSession(tssSign.exportSession());
-    } catch (err) {
-      if (err.message?.startsWith('TSS_ROUND_ALREADY_DONE')) {
-        const sig = await tssSign.getSignatureFromServer();
-        if (!sig) {
-          throw new Error('It looks like the TSS signature session was interrupted. Try deleting this proposal and creating a new one.');
+    let isTransientError = false;
+    let connErrs = 0;
+    do {
+      try {
+        await tssSign.start({
+          id,
+          messageHash,
+          derivationPath,
+          password
+        });
+        storeSession(tssSign.exportSession());
+      } catch (err) {
+        isTransientError = false; // reset
+        if (err.message?.startsWith('TSS_ROUND_ALREADY_DONE')) {
+          const sig = await tssSign.getSignatureFromServer();
+          if (!sig) {
+            throw new Error('It looks like the TSS signature session was interrupted. Try deleting this proposal and creating a new one.');
+          }
+          return {
+            signature: transformISignature(sig),
+            publicKey: sig.pubKey
+          };
+        } else if (err instanceof Errors.CONNECTION_ERROR) {
+          isTransientError = true;
+          connErrs = connResilience(connErrs, err);
+        } else {
+          throw err;
         }
-        return {
-          signature: transformISignature(sig),
-          publicKey: sig.pubKey
-        };
       }
-      throw err;
-    }
+    } while (isTransientError);
   }
 
   const spinner = prompt.spinner({ indicator: 'timer', onCancel: () => { tssSign.unsubscribe(); } });
@@ -87,8 +105,10 @@ export async function sign(args: {
     let rejected = false;
     const reject = (err) => { if (!rejected) { rejected = true; _reject(err); } };
 
+    let connErrs = 0;
     tssSign.subscribe();
     tssSign.on('roundsubmitted', (round) => {
+      connErrs = 0;
       storeSession(tssSign.exportSession());
       spinner.message(`Round ${round} submitted`);
     });
@@ -103,8 +123,13 @@ export async function sign(args: {
         spinner.cancel(e.message);
         rmSessionState();
         return reject(new ProcessCancelled());
+      } else if (e instanceof Errors.CONNECTION_ERROR) {
+        // Reduce the noise of transient errors
+        connErrs = connResilience(connErrs, e);
+        return;
+      } else {
+        prompt.log.error('Unexpected error during TSS signing: ' + (e.stack || e));
       }
-      prompt.log.error('Unexpected error during TSS signing: ' + (e.stack || e));
     });
     tssSign.on('complete', async () => {
       try {

packages/bitcore-wallet-client/src/lib/errors/spec.ts

@@ -147,7 +147,7 @@ export const errorSpec: IErrorSpec[] = [
       'We were unable to parse your payment. Please try again or contact your wallet provider.'
   },
   {
-    name: 'NO_TRASACTION',
+    name: 'NO_TRANSACTION',
     message:
       'Your request did not include a transaction. Please try again or contact your wallet provider.'
   },

packages/bitcore-wallet-client/src/lib/payproV2.ts

@@ -118,7 +118,7 @@ export class PayProV2 {
       case errMsg.includes('We were unable to parse your payment.'):
         return new Errors.UNABLE_TO_PARSE_PAYMENT();
       case errMsg.includes('Request must include exactly one'):
-        return new Errors.NO_TRASACTION();
+        return new Errors.NO_TRANSACTION();
       case errMsg.includes('Your transaction was an in an invalid format'):
         return new Errors.INVALID_TX_FORMAT();
       case errMsg.includes('We were unable to parse the transaction you sent'):

packages/bitcore-wallet-client/test/tss.test.ts

@@ -4,6 +4,10 @@ import sinon from 'sinon';
 import * as chai from 'chai';
 import BWS from '@bitpay-labs/bitcore-wallet-service';
 import { Defaults as BwsDefaults } from '@bitpay-labs/bitcore-wallet-service/ts_build/src/lib/common/defaults';
+import {
+  TssKeyGen as BwsTssKeyGen,
+  TssSign as BwsTssSign
+} from '@bitpay-labs/bitcore-wallet-service/ts_build/src/lib/tss';
 import request from 'supertest';
 import crypto from 'crypto';
 import fs from 'fs';
@@ -71,6 +75,17 @@ describe('TSS', function() {
     });
   });
 
+  beforeEach(function() {
+    sandbox.stub(BwsTssKeyGen, 'getMessagesForParty').callsFake(async function(params) {
+      params.maxWaitTimeSec = 1;
+      return (BwsTssKeyGen.getMessagesForParty as any).wrappedMethod.call(this, params);
+    });
+    sandbox.stub(BwsTssSign, 'getMessagesForParty').callsFake(async function(params) {
+      params.maxWaitTimeSec = 1;
+      return (BwsTssSign.getMessagesForParty as any).wrappedMethod.call(this, params);
+    });
+  });
+
   after(function(done) {
     dbConnection.close(done);
   });

packages/bitcore-wallet-service/README.md

@@ -23,7 +23,7 @@ More about BWS at https://blog.bitpay.com/announcing-the-bitcore-wallet-suite/
 ```sh
 git clone https://github.com/bitpay/bitcore.git
 cd bitcore
-npm install
+npm ci
 npm run bws
 ```
 
@@ -86,9 +86,6 @@ There are plenty examples of creating and sending proposals in the `/test/integr
 - copay running on port: 8100
 - bitcoin-core running on regtest mode (blue icon logo)
 
-> mongo topology crashes sometimes due to notifications being incompatible in a web browser
-> **bitcore-wallet-service/lib/notificationbroadcaster.js**
-> Note: If testing on a PC browser, comment out notificationbroadcaster.js to disable notifications.
 
 ### Steps:
 

packages/bitcore-wallet-service/src/lib/messagebroker.ts

@@ -2,10 +2,12 @@ import { EventEmitter } from 'events';
 import * as io from 'socket.io-client';
 import 'source-map-support/register';
 import logger from './logger';
+import type { Notification } from './model/notification';
 
 export class MessageBroker extends EventEmitter {
-  remote: boolean;
+  remote: boolean = false;
   mq: io.Socket;
+
   constructor(opts) {
     super();
 
@@ -28,15 +30,19 @@ export class MessageBroker extends EventEmitter {
     }
   }
 
-  send(data) {
+  send(data: Notification) {
     if (this.remote) {
       this.mq.emit('msg', data);
     } else {
       this.emit('msg', data);
     }
   }
 
-  onMessage(handler) {
+  onMessage(handler: (data: Notification) => void) {
     this.on('msg', handler);
   }
+
+  offMessage(handler: (data: Notification) => void) {
+    this.off('msg', handler);
+  }
 }

packages/bitcore-wallet-service/src/lib/model/notification.ts

@@ -24,15 +24,15 @@
 export interface INotification {
   version: string;
   createdOn: number;
-  id: number;
+  id: string | number;
   type: string;
   data: any;
   walletId: string;
   creatorId: string;
   isCreator: boolean;
 }
 
-export class Notification {
+export class Notification implements INotification {
   version: string;
   createdOn: number;
   id: string | number;

packages/bitcore-wallet-service/src/lib/routes/helpers/error.ts

@@ -13,6 +13,7 @@ export class ApiErrorHelper {
   returnError(err: any, res: express.Response, req: express.Request): void {
     // make sure headers have not been sent as this leads to an uncaught error
     if (res.headersSent) {
+      res.end();
       return;
     }
     if (err instanceof ClientError) {

packages/bitcore-wallet-service/src/lib/routes/tss.ts

@@ -37,17 +37,33 @@ export class TssRouter {
     });
 
     router.get('/v1/tss/keygen/:id/:round', authTssRequest(), async function(req, res) {
+      let interval: NodeJS.Timeout;
       try {
         const { id, round } = req.params as { [key: string]: string };
+        const { maxWaitTime } = req.query as { [key: string]: string };
         const copayerId = req.headers['x-identity'];
         if (round === 'secret') {
           const secret = await TssKeyGen.getBwsJoinSecret({ id, copayerId });
           return res.json({ secret });
         }
-        const { messages, publicKey } = await TssKeyGen.getMessagesForParty({ id, round: parseInt(round), copayerId });
-        return res.json({ messages, publicKey });
+
+        // Validate access and fetch session before committing to a streaming response, so that errors like
+        //   "session not found" or "not a participant" can still be returned with a proper
+        //   HTTP status instead of silently becoming an empty 200 (see below).
+        const session = await TssKeyGen.getSessionForCopayer({ id, copayerId });
+
+        // Keep the connection alive while waiting for the change stream to return a result.
+        // Headers must be finalized before writing the first heartbeat byte.
+        res.writeHead(200, { 'Content-Type': 'application/json' });
+        interval = setInterval(() => res.write('\n'), 1000);
+        req.on('close', () => clearInterval(interval));
+
+        const { messages, publicKey } = await TssKeyGen.getMessagesForParty({ session, round: parseInt(round), copayerId, maxWaitTimeSec: parseInt(maxWaitTime) });
+        return res.end(JSON.stringify({ messages, publicKey }));
       } catch (err) {
         return returnError(err ?? 'unknown', res, req);
+      } finally {
+        clearInterval(interval);
       }
     });
 
@@ -104,13 +120,30 @@ export class TssRouter {
     });
 
     router.get('/v1/tss/sign/:id/:round', authTssRequest(), async function(req, res) {
+      let interval: NodeJS.Timeout;
       try {
         const { id, round } = req.params as { [key: string]: string };
+        const { maxWaitTime } = req.query as { [key: string]: string };
         const copayerId = req.headers['x-identity'];
-        const { messages, signature, participants } = await TssSign.getMessagesForParty({ id, round: parseInt(round), copayerId });
-        return res.json({ messages, signature, participants });
+
+        // Validate access and fetch session before committing to a streaming response, so that errors like
+        //   "session not found" or "not a participant" can still be returned with a proper
+        //   HTTP status instead of silently becoming an empty 200 (see below).
+        const session = await TssSign.getSessionForCopayer({ id, copayerId });
+
+        // Keep the connection alive while waiting for the change stream to return a result.
+        // Headers must be finalized before writing the first heartbeat byte.
+        res.writeHead(200, { 'Content-Type': 'application/json' });
+        interval = setInterval(() => res.write('\n'), 1000);
+        req.on('close', () => clearInterval(interval));
+
+        const { messages, signature, participants } = await TssSign.getMessagesForParty({ session, round: parseInt(round), copayerId, maxWaitTimeSec: parseInt(maxWaitTime) });
+        clearInterval(interval);
+        return res.end(JSON.stringify({ messages, signature, participants }));
       } catch (err) {
         return returnError(err ?? 'unknown', res, req);
+      } finally {
+        clearInterval(interval);
       }
     });
 

packages/bitcore-wallet-service/src/lib/server.ts

@@ -326,12 +326,9 @@ export class WalletService implements IWalletService {
     );
   }
 
-  static handleIncomingNotifications(notification, cb) {
-    cb = cb || function() { };
-
+  static handleIncomingNotifications(_notification: INotification) {
     // do nothing here....
     // bc height cache is cleared on bcmonitor
-    return cb();
   }
 
   static shutDown(cb) {
@@ -474,6 +471,13 @@ export class WalletService implements IWalletService {
     return storage;
   }
 
+  static getMessageBroker() {
+    if (!initialized) {
+      throw new Error('Message broker requested before server was initialized');
+    }
+    return messageBroker;
+  }
+
   _runLocked(cb, task, waitTime?: number) {
     $.checkState(this.walletId, 'Failed state: this.walletId undefined at <_runLocked()>');
     this.lock.runLocked(this.walletId, { waitTime }, cb, task);
@@ -1152,7 +1156,7 @@ export class WalletService implements IWalletService {
         try {
           const isValid = this._verifyRequestPubKey(opts.requestPubKey, opts.signature, target.xPubKey);
           if (!isValid) return cb(Errors.NOT_AUTHORIZED);
-        } catch (e) {
+        } catch {
           return cb(Errors.NOT_AUTHORIZED);
         }
 
@@ -3850,7 +3854,7 @@ export class WalletService implements IWalletService {
           const notifications = res
             .flat()
             .map((n: INotification) => ({ ...n, walletId: this.walletId }))
-            .sort((a, b) => a.id - b.id);
+            .sort((a, b) => a.id?.toString()?.localeCompare(b.id?.toString()));
 
           return cb(null, notifications);
         }

packages/bitcore-wallet-service/src/lib/storage.ts

@@ -1,6 +1,5 @@
 import * as async from 'async';
 import _ from 'lodash';
-import { Db } from 'mongodb';
 import * as mongodb from 'mongodb';
 import preconditions from 'preconditions';
 import { BCHAddressTranslator } from './bchaddresstranslator'; // only for migration
@@ -19,8 +18,11 @@ import {
   TxProposal,
   Wallet
 } from './model';
-import { ITssKeyMessageObject, TssKeyGenModel } from './model/tsskeygen';
-import { ITssSigMessageObject, TssSigGenModel } from './model/tsssign';
+import { TssKeyGenModel } from './model/tsskeygen';
+import { TssSigGenModel } from './model/tsssign';
+import type { ITssKeyMessageObject } from './model/tsskeygen';
+import type { ITssSigMessageObject } from './model/tsssign';
+import type { Db } from 'mongodb';
 
 const $ = preconditions.singleton();