fix: waitForTransactionToBeProvable not waiting for blocks

lib/externalApis/tenderdash/BlockchainListener.js

@@ -1,3 +1,5 @@
+const crypto = require('crypto');
+
 const EventEmitter = require('events');
 const TransactionWaitPeriodExceededError = require('../../errors/TransactionWaitPeriodExceededError');
 
@@ -26,6 +28,15 @@ class BlockchainListener extends EventEmitter {
     return `transaction:${transactionHashString}`;
   }
 
+  /**
+   *
+   * @param transactionHashString
+   * @return {string}
+   */
+  static getTransactionAddedToTheBlockEventName(transactionHashString) {
+    return `blockTransactionAdded:${transactionHashString}`;
+  }
+
   /**
    * Subscribe to transactions and attach transaction event handler
    */
@@ -36,6 +47,7 @@ class BlockchainListener extends EventEmitter {
     this.wsClient.subscribe(NEW_BLOCK_QUERY);
     this.wsClient.on(NEW_BLOCK_QUERY, (message) => {
       this.emit(events.NEW_BLOCK, message);
+      this.emitTransactionHashesInBlock(message.data.value.block.data.txs);
     });
   }
 
@@ -55,6 +67,16 @@ class BlockchainListener extends EventEmitter {
     return handler;
   }
 
+  emitTransactionHashesInBlock(serializedTransactions) {
+    serializedTransactions.forEach((base64tx) => {
+      const hashString = crypto.createHash('sha256')
+        .update(Buffer.from(base64tx, 'base64'))
+        .digest()
+        .toString('hex');
+      this.emit(BlockchainListener.getTransactionAddedToTheBlockEventName(hashString));
+    });
+  }
+
   /**
    * Emits transaction:%tx_hash% if there's a transaction in the message
    * @private
@@ -76,37 +98,56 @@ class BlockchainListener extends EventEmitter {
    * @param {number} [timeout] - timeout to reject after
    * @return {Promise<Object>}
    */
-  waitForTransaction(hashString, timeout = 60000) {
+  waitForTransactionToBeProvable(hashString, timeout = 60000) {
     const topic = BlockchainListener.getTransactionEventName(hashString);
-    let handler;
+    const txInBlockTopic = BlockchainListener
+      .getTransactionAddedToTheBlockEventName(hashString.toLowerCase());
+
+
+    let txHandler;
+    const txPromise = new Promise(((resolve) => {
+      txHandler = this.createPromiseHandler(topic, resolve);
+      this.on(topic, txHandler);
+    }));
+
+    let txInBlockHandler;
+    let newBlockHandler;
+    const txInBlockPromise = new Promise(((resolve) => {
+      let seenTransaction = false;
+      txInBlockHandler = () => {
+        seenTransaction = true;
+      };
+      newBlockHandler = () => {
+        if (seenTransaction) {
+          // Note that this will resolve only after two blocks. That is because the first block will
+          // flip the 'seenTransaction' toggle to true, and transaction will become provable
+          // only on the next block after the block it was included into
+          this.off(events.NEW_BLOCK, newBlockHandler);
+          resolve();
+        }
+      };
+      this.once(txInBlockTopic, txInBlockHandler);
+      this.on(events.NEW_BLOCK, newBlockHandler);
+    }));
 
     return Promise.race([
-      new Promise((resolve) => {
-        handler = this.createPromiseHandler(topic, resolve);
-        this.on(topic, handler);
+      new Promise(async (resolve) => {
+        Promise.all([
+          txPromise,
+          txInBlockPromise,
+        ]).then(([data]) => resolve(data));
       }),
       new Promise((resolve, reject) => {
         setTimeout(() => {
-          this.off(topic, handler);
+          // Detaching old handlers
+          this.off(topic, txHandler);
+          this.off(txInBlockTopic, txInBlockHandler);
+          this.off(events.NEW_BLOCK, newBlockHandler);
           reject(new TransactionWaitPeriodExceededError(hashString));
         }, timeout);
       }),
     ]);
   }
-
-  waitForNextBlock() {
-    return new Promise((resolve) => {
-      this.once(events.NEW_BLOCK, resolve);
-    });
-  }
-
-  async waitForBlocks(countToWait = 1) {
-    let blocksSeen = 0;
-    while (blocksSeen !== countToWait) {
-      await this.waitForNextBlock();
-      blocksSeen += 1;
-    }
-  }
 }
 
 BlockchainListener.TX_QUERY = TX_QUERY;

lib/grpcServer/handlers/platform/waitForStateTransitionResultHandlerFactory.js

@@ -78,7 +78,8 @@ function waitForStateTransitionResultHandlerFactory(
     let data;
 
     try {
-      data = await blockchainListener.waitForTransaction(hashString, stateTransitionWaitTimeout);
+      data = await blockchainListener
+        .waitForTransactionToBeProvable(hashString, stateTransitionWaitTimeout);
     } catch (e) {
       if (e instanceof TransactionWaitPeriodExceededError) {
         throw new DeadlineExceededGrpcError(
@@ -114,10 +115,6 @@ function waitForStateTransitionResultHandlerFactory(
       return response;
     }
 
-    // The first block is the one with the current ST, second block is the
-    // block in which st result will be included in the proof
-    await blockchainListener.waitForBlocks(2);
-
     if (prove) {
       const stateTransition = await dpp.stateTransition.createFromBuffer(
         Buffer.from(tx, 'base64'),

test/integration/externalApis/tenderdash/BlockchainListener.spec.js

@@ -1,4 +1,5 @@
 const EventEmitter = require('events');
+const crypto = require('crypto');
 const BlockchainListener = require('../../../../lib/externalApis/tenderdash/BlockchainListener');
 const TransactionWaitPeriodExceededError = require('../../../../lib/errors/TransactionWaitPeriodExceededError');
 
@@ -7,6 +8,10 @@ describe('BlockchainListener', () => {
   let wsClientMock;
   let blockchainListener;
   let txDataMock;
+  const base64tx = 'aaaa';
+  let txHash;
+  let blockWithTxMock;
+  let emptyBlockMock;
 
   beforeEach(function beforeEach() {
     ({ sinon } = this);
@@ -15,12 +20,24 @@ describe('BlockchainListener', () => {
     blockchainListener = new BlockchainListener(wsClientMock);
     blockchainListener.start();
 
+    txHash = crypto.createHash('sha256')
+      .update(Buffer.from(base64tx, 'base64'))
+      .digest()
+      .toString('hex');
+
     txDataMock = {
       events: {
-        'tx.hash': ['123'],
+        'tx.hash': [txHash],
       },
     };
 
+    blockWithTxMock = {
+      data: { value: { block: { data: { txs: [base64tx] } } } },
+    };
+    emptyBlockMock = {
+      data: { value: { block: { data: { txs: [] } } } },
+    };
+
     sinon.spy(blockchainListener, 'on');
     sinon.spy(blockchainListener, 'off');
     sinon.spy(blockchainListener, 'emit');
@@ -40,47 +57,58 @@ describe('BlockchainListener', () => {
 
   describe('.getTransactionEventName', () => {
     it('should return transaction event name', () => {
-      const eventName = BlockchainListener.getTransactionEventName('123');
-      expect(eventName).to.be.equal('transaction:123');
+      const eventName = BlockchainListener.getTransactionEventName(txHash);
+      expect(eventName).to.be.equal(`transaction:${txHash}`);
     });
   });
 
   describe('#waitForTransaction', () => {
     it('should remove listener after transaction resolves', async () => {
-      const eventName = BlockchainListener.getTransactionEventName('123');
-      const txPromise = blockchainListener.waitForTransaction('123', 2000);
+      const txPromise = blockchainListener.waitForTransactionToBeProvable(txHash, 2000);
 
-      expect(blockchainListener.listenerCount(eventName)).to.be.equal(1);
+      // check that we attached events correctly
+      const events = blockchainListener.eventNames();
+      // transaction result, transaction in block an block events
+      expect(events.length).to.be.equal(3);
+      events.forEach((eventName) => {
+        expect(blockchainListener.listenerCount(eventName)).to.be.equal(1);
+      });
 
       setTimeout(() => {
         wsClientMock.emit(BlockchainListener.TX_QUERY, Object.assign({}, txDataMock));
+        wsClientMock.emit(BlockchainListener.NEW_BLOCK_QUERY, Object.assign({}, blockWithTxMock));
+        wsClientMock.emit(BlockchainListener.NEW_BLOCK_QUERY, Object.assign({}, emptyBlockMock));
       }, 10);
 
       const txData = await txPromise;
 
       // Check that event listener was properly attached
-      expect(blockchainListener.on).to.be.calledOnce();
-      // Check that transaction data was emitted
-      expect(blockchainListener.emit).to.be.calledOnce();
+      expect(blockchainListener.on).to.be.calledThrice();
       // Check that the event listener was properly removed
-      expect(blockchainListener.off).to.be.calledOnce();
-      expect(blockchainListener.listenerCount(eventName)).to.be.equal(0);
+      expect(blockchainListener.off).to.be.calledTwice();
+      events.forEach((eventName) => {
+        expect(blockchainListener.listenerCount(eventName)).to.be.equal(0);
+      });
 
       expect(txData).to.be.deep.equal(txDataMock);
     });
 
     it('should not emit transaction event if event data has no transaction', async () => {
-      const eventName = BlockchainListener.getTransactionEventName('123');
-      txDataMock = {};
-
+      let events;
       let error;
       try {
-        const txPromise = blockchainListener.waitForTransaction('123', 1000);
+        const txPromise = blockchainListener.waitForTransactionToBeProvable(txHash, 1000);
 
-        expect(blockchainListener.listenerCount(eventName)).to.be.equal(1);
+        // check that we attached events correctly
+        events = blockchainListener.eventNames();
+        // transaction result, transaction in block an block events
+        expect(events.length).to.be.equal(3);
+        events.forEach((eventName) => {
+          expect(blockchainListener.listenerCount(eventName)).to.be.equal(1);
+        });
 
         setTimeout(() => {
-          wsClientMock.emit(BlockchainListener.TX_QUERY, Object.assign({}, txDataMock));
+          wsClientMock.emit(BlockchainListener.NEW_BLOCK_QUERY, Object.assign({}, blockWithTxMock));
         }, 10);
 
         await txPromise;
@@ -90,64 +118,50 @@ describe('BlockchainListener', () => {
 
       // Check that the error is correct
       expect(error).to.be.instanceOf(TransactionWaitPeriodExceededError);
-      expect(error.message).to.be.equal('Transaction waiting period for 123 exceeded');
-      expect(error.getTransactionHash()).to.be.equal('123');
+      expect(error.message).to.be.equal(`Transaction waiting period for ${txHash} exceeded`);
+      expect(error.getTransactionHash()).to.be.equal(txHash);
 
       // Check that event listener was properly attached
-      expect(blockchainListener.on).to.be.calledOnce();
-      // Check that event listener was properly removed
-      expect(blockchainListener.off).to.be.calledOnce();
-      expect(blockchainListener.listenerCount(eventName)).to.be.equal(0);
+      expect(blockchainListener.on).to.be.calledThrice();
+      events.forEach((eventName) => {
+        expect(blockchainListener.listenerCount(eventName)).to.be.equal(0);
+      });
       // Check that no transaction data was emitted
-      expect(blockchainListener.emit).to.not.be.called();
+      expect(blockchainListener.emit).to.be.calledTwice();
     });
 
     it('should remove listener after timeout has been exceeded', async () => {
-      const eventName = BlockchainListener.getTransactionEventName('123');
+      let events;
       let error;
       try {
-        await blockchainListener.waitForTransaction('123', 100);
+        const txPromise = blockchainListener.waitForTransactionToBeProvable(txHash, 100);
+
+        events = blockchainListener.eventNames();
+        expect(events.length).to.be.equal(3);
+        events.forEach((eventName) => {
+          expect(blockchainListener.listenerCount(eventName)).to.be.equal(1);
+        });
+
+        await txPromise;
       } catch (e) {
         error = e;
       }
 
       // Check that the error is correct
       expect(error).to.be.instanceOf(TransactionWaitPeriodExceededError);
-      expect(error.message).to.be.equal('Transaction waiting period for 123 exceeded');
-      expect(error.getTransactionHash()).to.be.equal('123');
+      expect(error.message).to.be.equal(`Transaction waiting period for ${txHash} exceeded`);
+      expect(error.getTransactionHash()).to.be.equal(txHash);
 
       // Check that event listener was properly attached
-      expect(blockchainListener.on).to.be.calledOnce();
+      expect(blockchainListener.on).to.be.calledThrice();
       // Check that event listener was properly removed
-      expect(blockchainListener.off).to.be.calledOnce();
-      expect(blockchainListener.listenerCount(eventName)).to.be.equal(0);
+      expect(blockchainListener.off).to.be.calledThrice();
+
+      events.forEach((eventName) => {
+        expect(blockchainListener.listenerCount(eventName)).to.be.equal(0);
+      });
       // Check that no transaction data was emitted
       expect(blockchainListener.emit).to.not.be.called();
     });
   });
-
-  describe('#waitForBlocks', () => {
-    it('should wait for n blocks and remove listeners afterwards', async () => {
-      const newBlockEvent = BlockchainListener.events.NEW_BLOCK;
-      const blockPromise = blockchainListener.waitForBlocks(2);
-
-      expect(blockchainListener.listenerCount(newBlockEvent)).to.be.equal(1);
-
-      setTimeout(() => {
-        wsClientMock.emit(BlockchainListener.NEW_BLOCK_QUERY, Object.assign({}, txDataMock));
-      }, 10);
-      setTimeout(() => {
-        wsClientMock.emit(BlockchainListener.NEW_BLOCK_QUERY, Object.assign({}, txDataMock));
-      }, 10);
-
-      await blockPromise;
-
-      // Check that event listener was properly attached
-      expect(blockchainListener.on).to.be.calledTwice();
-      // Check that transaction data was emitted
-      expect(blockchainListener.emit).to.be.calledTwice();
-      // Check that the event listener was properly removed
-      expect(blockchainListener.listenerCount(newBlockEvent)).to.be.equal(0);
-    });
-  });
 });