Merge pull request #3965 from LiskHQ/3962-transactions-not-being-broa…

…dcasted

Some transaction events are not being broadcasted - Closes #3962

framework/src/modules/chain/blocks/blocks.js

@@ -327,6 +327,7 @@ class Blocks extends EventEmitter {
 					await this._updateBroadhash();
 					this._lastBlock = newBlock;
 					this._isActive = false;
+					this.emit(EVENT_NEW_BLOCK, { block: cloneDeep(this._lastBlock) });
 				} catch (error) {
 					this._isActive = false;
 					this.logger.error(error);
@@ -417,6 +418,7 @@ class Blocks extends EventEmitter {
 						validBlock => this.broadcast(validBlock)
 					);
 					await this._updateBroadhash();
+					this.emit(EVENT_NEW_BLOCK, { block: cloneDeep(this._lastBlock) });
 					this._isActive = false;
 				} catch (error) {
 					this.logger.error(error);

framework/src/modules/chain/chain.js

@@ -27,7 +27,11 @@ const { bootstrapStorage, bootstrapCache } = require('./init_steps');
 const jobQueue = require('./utils/jobs_queue');
 const { Peers } = require('./peers');
 const { TransactionInterfaceAdapter } = require('./interface_adapters');
-const { TransactionPool } = require('./transaction_pool');
+const {
+	TransactionPool,
+	EVENT_MULTISIGNATURE_SIGNATURE,
+	EVENT_UNCONFIRMED_TRANSACTION,
+} = require('./transaction_pool');
 const { Rounds } = require('./rounds');
 const {
 	BlockSlots,
@@ -505,15 +509,33 @@ module.exports = class Chain {
 			this.channel.publish('chain:blocks:change', block);
 		});
 
+		this.transactionPool.on(EVENT_UNCONFIRMED_TRANSACTION, transaction => {
+			this.logger.trace(
+				{ transactionId: transaction.id },
+				'Received EVENT_UNCONFIRMED_TRANSACTION'
+			);
+			this.transport.onUnconfirmedTransaction(transaction, true);
+		});
+
 		this.blocks.on(EVENT_NEW_BROADHASH, ({ broadhash, height }) => {
 			this.channel.invoke('app:updateApplicationState', { broadhash, height });
 		});
+
+		this.transactionPool.on(EVENT_MULTISIGNATURE_SIGNATURE, signature => {
+			this.logger.trace(
+				{ signature },
+				'Received EVENT_MULTISIGNATURE_SIGNATURE'
+			);
+			this.transport.onSignature(signature, true);
+		});
 	}
 
 	_unsubscribeToEvents() {
 		this.blocks.removeAllListeners(EVENT_BROADCAST_BLOCK);
 		this.blocks.removeAllListeners(EVENT_DELETE_BLOCK);
 		this.blocks.removeAllListeners(EVENT_NEW_BLOCK);
 		this.blocks.removeAllListeners(EVENT_NEW_BROADHASH);
+		this.blocks.removeAllListeners(EVENT_UNCONFIRMED_TRANSACTION);
+		this.blocks.removeAllListeners(EVENT_MULTISIGNATURE_SIGNATURE);
 	}
 };

framework/src/modules/chain/transaction_pool/index.js

@@ -14,8 +14,14 @@
 
 'use strict';
 
-const { TransactionPool } = require('./transaction_pool');
+const {
+	TransactionPool,
+	EVENT_UNCONFIRMED_TRANSACTION,
+	EVENT_MULTISIGNATURE_SIGNATURE,
+} = require('./transaction_pool');
 
 module.exports = {
 	TransactionPool,
+	EVENT_UNCONFIRMED_TRANSACTION,
+	EVENT_MULTISIGNATURE_SIGNATURE,
 };

framework/src/modules/chain/transaction_pool/transaction_pool.js

@@ -25,6 +25,9 @@ const { getAddressFromPublicKey } = require('@liskhq/lisk-cryptography');
 const { sortBy } = require('./sort');
 const transactionsModule = require('../transactions');
 
+const EVENT_UNCONFIRMED_TRANSACTION = 'EVENT_UNCONFIRMED_TRANSACTION';
+const EVENT_MULTISIGNATURE_SIGNATURE = 'EVENT_MULTISIGNATURE_SIGNATURE';
+
 const handleAddTransactionResponse = (addTransactionResponse, transaction) => {
 	if (addTransactionResponse.isFull) {
 		throw new Error('Transaction pool is full');
@@ -160,7 +163,7 @@ class TransactionPool extends EventEmitter {
 		this.pool.on(pool.EVENT_VERIFIED_TRANSACTION_ONCE, ({ payload }) => {
 			if (payload.length > 0) {
 				payload.forEach(aTransaction =>
-					this.emit('unconfirmedTransaction', aTransaction)
+					this.emit(EVENT_UNCONFIRMED_TRANSACTION, aTransaction)
 				);
 			}
 		});
@@ -224,6 +227,8 @@ class TransactionPool extends EventEmitter {
 			this.logger.error(message, { signature });
 			throw transactionResponse.errors;
 		}
+
+		this.emit(EVENT_MULTISIGNATURE_SIGNATURE, signature);
 		return transactionResponse;
 	}
 
@@ -513,4 +518,8 @@ class TransactionPool extends EventEmitter {
 	}
 }
 
-module.exports = { TransactionPool };
+module.exports = {
+	TransactionPool,
+	EVENT_UNCONFIRMED_TRANSACTION,
+	EVENT_MULTISIGNATURE_SIGNATURE,
+};

framework/src/modules/chain/transport/broadcaster.js

@@ -257,7 +257,7 @@ class Broadcaster {
 		}
 		try {
 			await this.filterQueue();
-			const broadcasts = this.queue.splice(0, this.broadcasts.releaseLimit);
+			const broadcasts = this.queue.splice(0, this.config.releaseLimit);
 			const squashedBroadcasts = this.squashQueue(broadcasts);
 
 			await Promise.all(

framework/test/mocha/unit/modules/chain/blocks/blocks.js

@@ -232,6 +232,22 @@ describe('blocks', () => {
 
 					expect(blocksInstance.blocksProcess.processBlock).to.be.calledOnce;
 				});
+
+				it('should emit EVENT_NEW_BLOCK with block', async () => {
+					const emitSpy = sinonSandbox.spy(blocksInstance, 'emit');
+					const fakeBlock = {
+						id: '5',
+						previousBlock: '2',
+						height: 3,
+					};
+					await blocksInstance.receiveBlockFromNetwork(fakeBlock);
+
+					expect(blocksInstance.blocksProcess.processBlock).to.be.calledOnce;
+					expect(emitSpy.secondCall.args).to.be.eql([
+						'EVENT_NEW_BLOCK',
+						{ block: fakeBlock },
+					]);
+				});
 			});
 
 			describe('when block.previousBlock !== lastBlock.id && lastBlock.height + 1 === block.height', () => {
@@ -314,6 +330,23 @@ describe('blocks', () => {
 					expect(blocksInstance._isActive).to.be.false;
 					expect(loggerStub.warn).to.be.calledOnce;
 				});
+
+				it('should emit EVENT_NEW_BLOCK with block', async () => {
+					const emitSpy = sinonSandbox.spy(blocksInstance, 'emit');
+					const forkFiveBlock = {
+						id: '5',
+						generatorPublicKey: 'a',
+						previousBlock: '1',
+						height: 2,
+					};
+					await blocksInstance.receiveBlockFromNetwork(forkFiveBlock);
+
+					expect(blocksInstance.blocksProcess.processBlock).to.be.calledOnce;
+					expect(emitSpy.thirdCall.args).to.be.eql([
+						'EVENT_NEW_BLOCK',
+						{ block: forkFiveBlock },
+					]);
+				});
 			});
 
 			describe('when block.id === lastBlock.id', () => {

framework/test/mocha/unit/modules/chain/transaction_pool/transaction_pool.js

@@ -22,6 +22,8 @@ const {
 const { expect } = require('chai');
 const {
 	TransactionPool,
+	EVENT_UNCONFIRMED_TRANSACTION,
+	EVENT_MULTISIGNATURE_SIGNATURE,
 } = require('../../../../../../src/modules/chain/transaction_pool/transaction_pool');
 const transactionsModule = require('../../../../../../src/modules/chain/transactions');
 const { transactions: transactionsFixtures } = require('../../../../fixtures');
@@ -77,6 +79,10 @@ describe('transactionPool', () => {
 			maxTransactionsPerBlock,
 		});
 
+		// Stubs for event emitters
+		sinonSandbox.stub(transactionPool.pool, 'on');
+		sinonSandbox.stub(transactionPool, 'emit');
+
 		dummyTransactions = [{ id: 1 }, { id: 2 }, { id: 3 }];
 	});
 
@@ -435,6 +441,30 @@ describe('transactionPool', () => {
 				}
 			});
 		});
+
+		describe('events', () => {
+			beforeEach(async () => {
+				sinonSandbox.stub(transactionsModule, 'processSignature').returns(
+					sinonSandbox.stub().resolves({
+						...transactionResponse,
+						status: 2,
+						errors: [],
+					})
+				);
+			});
+
+			it('should emit EVENT_MULTISIGNATURE_SIGNATURE', async () => {
+				// Act
+				await transactionPool.getTransactionAndProcessSignature(
+					signatureObject
+				);
+				// Assert
+				expect(transactionPool.emit).to.be.calledWith(
+					EVENT_MULTISIGNATURE_SIGNATURE,
+					signatureObject
+				);
+			});
+		});
 	});
 
 	describe('processUnconfirmedTransaction', () => {
@@ -519,4 +549,36 @@ describe('transactionPool', () => {
 			}
 		});
 	});
+
+	describe('#Events', () => {
+		it('it should subscribe to EVENT_VERIFIED_TRANSACTION_ONCE, EVENT_ADDED_TRANSACTIONS, EVENT_REMOVED_TRANSACTIONS', async () => {
+			// Act
+			transactionPool.subscribeEvents();
+			// Assert
+			expect(transactionPool.pool.on.firstCall).to.be.calledWith(
+				'transactionVerifiedOnce'
+			);
+			expect(transactionPool.pool.on.secondCall).to.be.calledWith(
+				'transactionsAdded'
+			);
+			expect(transactionPool.pool.on.thirdCall).to.be.calledWith(
+				'transactionsRemoved'
+			);
+		});
+
+		it('should emit EVENT_UNCONFIRMED_TRANSACTION on EVENT_VERIFIED_TRANSACTION_ONCE', async () => {
+			// Arrange
+			const eventData = {
+				action: 'transactionVerifiedOnce',
+				payload: [dummyTransactions[0]],
+			};
+			// Act
+			transactionPool.pool.emit('transactionVerifiedOnce', eventData);
+			// Assert
+			expect(transactionPool.emit).to.be.calledWith(
+				EVENT_UNCONFIRMED_TRANSACTION,
+				dummyTransactions[0]
+			);
+		});
+	});
 });