Merge pull request #4211 from LiskHQ/4208-fix_event_handling

Fix event handling - Closes #4208

framework/src/modules/chain/chain.js

@@ -170,21 +170,30 @@ module.exports = class Chain {
 
 			// Avoid receiving blocks/transactions from the network during snapshotting process
 			if (!this.options.loading.rebuildUpToRound) {
-				this.channel.subscribe('network:event', ({ data: { event, data } }) => {
-					if (event === 'postTransactions') {
-						this.transport.postTransactions(data);
-						return;
-					}
-					if (event === 'postSignatures') {
-						this.transport.postSignatures(data);
-						return;
-					}
-					if (event === 'postBlock') {
-						this.transport.postBlock(data);
-						// eslint-disable-next-line no-useless-return
-						return;
-					}
-				});
+				this.channel.subscribe(
+					'network:event',
+					async ({ data: { event, data } }) => {
+						try {
+							if (event === 'postTransactions') {
+								await this.transport.postTransactions(data);
+								return;
+							}
+							if (event === 'postSignatures') {
+								await this.transport.postSignatures(data);
+								return;
+							}
+							if (event === 'postBlock') {
+								await this.transport.postBlock(data);
+								return;
+							}
+						} catch (error) {
+							this.logger.warn(
+								{ error, event },
+								'Received invalid event message',
+							);
+						}
+					},
+				);
 			}
 		} catch (error) {
 			this.logger.fatal('Chain initialization', {

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

@@ -344,13 +344,12 @@ class Transport {
 	 * @todo Add @returns tag
 	 * @todo Add description of the function
 	 */
-	async postBlock(query) {
+	async postBlock(query = {}) {
 		if (!this.constants.broadcasts.active) {
 			return this.logger.debug(
 				'Receiving blocks disabled by user through config.json',
 			);
 		}
-		query = query || {};
 
 		const errors = validator.validate(definitions.WSBlocksBroadcast, query);
 
@@ -363,39 +362,24 @@ class Transport {
 					query,
 				},
 			);
-			throw new Error(errors);
+			// TODO: If there is an error, invoke the applyPenalty action on the Network module once it is implemented.
+			throw errors;
 		}
 
-		let block;
-		let success = true;
-		try {
-			block = blocksUtils.addBlockProperties(query.block);
-
-			// Instantiate transaction classes
-			block.transactions = this.interfaceAdapters.transactions.fromBlock(block);
+		let block = blocksUtils.addBlockProperties(query.block);
 
-			block = blocksUtils.objectNormalize(block);
-		} catch (e) {
-			success = false;
-			this.logger.debug('Block normalization failed', {
-				err: e.toString(),
-				module: 'transport',
-				block: query.block,
-			});
+		// Instantiate transaction classes
+		block.transactions = this.interfaceAdapters.transactions.fromBlock(block);
 
-			// TODO: If there is an error, invoke the applyPenalty action on the Network module once it is implemented.
-		}
+		block = blocksUtils.objectNormalize(block);
 		// TODO: endpoint should be protected before
 		if (this.loaderModule.syncing()) {
 			return this.logger.debug(
 				"Client is syncing. Can't receive block at the moment.",
 				block.id,
 			);
 		}
-		if (success) {
-			return this.blocksModule.receiveBlockFromNetwork(block);
-		}
-		return null;
+		return this.blocksModule.receiveBlockFromNetwork(block);
 	}
 
 	/**
@@ -449,6 +433,7 @@ class Transport {
 
 		if (errors.length) {
 			this.logger.debug('Invalid signatures body', errors);
+			// TODO: If there is an error, invoke the applyPenalty action on the Network module once it is implemented.
 			throw errors;
 		}
 
@@ -543,6 +528,7 @@ class Transport {
 
 		if (errors.length) {
 			this.logger.debug('Invalid transactions body', errors);
+			// TODO: If there is an error, invoke the applyPenalty action on the Network module once it is implemented.
 			throw errors;
 		}
 

framework/test/mocha/functional/ws/transport/transport.js

@@ -32,7 +32,7 @@ describe('WS transport', () => {
 		let p2p;
 
 		function postTransaction(transactionToPost) {
-			p2p.send({
+			return p2p.send({
 				event: 'postTransactions',
 				data: {
 					nonce: 'sYHEDBKcScaAAAYg',
@@ -72,6 +72,10 @@ describe('WS transport', () => {
 				goodTransactions.push(transaction);
 				done();
 			});
+
+			it('should not crash the application', async () => {
+				await postTransaction('Invalid transaction');
+			});
 		});
 
 		describe('confirmation', () => {
@@ -379,6 +383,22 @@ describe('WS transport', () => {
 				});
 				await p2p.send({ event: 'postBlock', data: { block: testBlock } });
 			});
+
+			it('should not crash the application when sending the invalid block', async () => {
+				await p2p.send({
+					event: 'postBlock',
+					data: { block: { generatorPublicKey: '123' } },
+				});
+			});
+		});
+
+		describe('postSignatures', () => {
+			it('should not crash the application when sending the invalid signatures', async () => {
+				await p2p.send({
+					event: 'postSignatures',
+					data: { signatures: ['not object signature'] },
+				});
+			});
 		});
 	});
 });

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

@@ -895,19 +895,15 @@ describe('transport', () => {
 							beforeEach(async () => {
 								sinonSandbox
 									.stub(blocksModule, 'objectNormalize')
-									.throws(blockValidationError);
-								transportModule.postBlock(postBlockQuery);
+									.throws(new Error(blockValidationError));
 							});
 
-							it('should call transportModule.logger.debug with "Block normalization failed" and {err: error, module: "transport", block: query.block }', async () => {
-								expect(transportModule.logger.debug).to.be.calledWith(
-									'Block normalization failed',
-									{
-										err: blockValidationError.toString(),
-										module: 'transport',
-										block: blockMock,
-									},
-								);
+							it('should throw an error', async () => {
+								try {
+									await transportModule.postBlock(postBlockQuery);
+								} catch (err) {
+									expect(err.message).to.equal(blockValidationError);
+								}
 							});
 						});