Improve block propagation - Closes #541

logic/broadcaster.js

@@ -5,6 +5,9 @@ var constants = require('../helpers/constants.js');
 var jobsQueue = require('../helpers/jobsQueue.js');
 var extend = require('extend');
 var _ = require('lodash');
+var BSON = require('bson');
+
+var bson = new BSON();
 
 // Private fields
 var modules, library, self, __private = {};
@@ -151,6 +154,9 @@ Broadcaster.prototype.broadcast = function (params, options, cb) {
 			}
 		},
 		function getFromPeer (peers, waterCb) {
+			if (options.data.block) {
+				options.data.block = bson.deserialize(options.data.block);
+			}
 			library.logger.debug('Begin broadcast', options);
 
 			if (params.limit === self.config.peerLimit) { 

modules/blocks/chain.js

@@ -327,13 +327,12 @@ __private.applyTransaction = function (block, transaction, sender, cb) {
  * @method applyBlock
  * @emits  SIGTERM
  * @param  {Object}   block Full normalized block
- * @param  {boolean}  broadcast Indicator that block needs to be broadcasted
- * @param  {Function} cb Callback function
  * @param  {boolean}  saveBlock Indicator that block needs to be saved to database
+ * @param  {Function} cb Callback function
  * @return {Function} cb Callback function from params (through setImmediate)
  * @return {Object}   cb.err Error if occurred
  */
-Chain.prototype.applyBlock = function (block, broadcast, cb, saveBlock) {
+Chain.prototype.applyBlock = function (block, saveBlock, cb) {
 	// Prevent shutdown during database writes.
 	modules.blocks.isActive.set(true);
 
@@ -458,13 +457,11 @@ Chain.prototype.applyBlock = function (block, broadcast, cb, saveBlock) {
 					}
 
 					library.logger.debug('Block applied correctly with ' + block.transactions.length + ' transactions');
-					library.bus.message('newBlock', block, broadcast);
 
 					// DATABASE write. Update delegates accounts
 					modules.rounds.tick(block, seriesCb);
 				});
 			} else {
-				library.bus.message('newBlock', block, broadcast);
 
 				// DATABASE write. Update delegates accounts
 				modules.rounds.tick(block, seriesCb);
@@ -497,6 +494,20 @@ Chain.prototype.applyBlock = function (block, broadcast, cb, saveBlock) {
 	});
 };
 
+/**
+ * Broadcast reduced block to increase network performance.
+ * @param {Object} reducedBlock reduced block
+ * @param {boolean} broadcast Indicator that block needs to be broadcasted
+ * @throws {string} Error description
+ */
+Chain.prototype.broadcastReducedBlock = function (reducedBlock, broadcast) {
+	try {
+		library.bus.message('newBlock', reducedBlock, broadcast);
+	} catch (e) {
+		throw 'reduced newBlock broadcast message error: ' + e;
+	}
+	library.logger.debug(['reducedBlock', reducedBlock.id, 'broadcasted correctly'].join(' '));
+};
 
 /**
  * Deletes last block, undo transactions, recalculate round

modules/blocks/process.js

@@ -179,10 +179,10 @@ Process.prototype.loadBlocksOffset = function (limit, offset, verify, cb) {
 				if (block.id === library.genesisblock.block.id) {
 					modules.blocks.chain.applyGenesisBlock(block, cb);
 				} else {
-					// Apply block - broadcast: false, saveBlock: false
+					// Apply block - saveBlock: false
 					// FIXME: Looks like we are missing some validations here, because applyBlock is different than processBlock used elesewhere
 					// - that need to be checked and adjusted to be consistent
-					modules.blocks.chain.applyBlock(block, false, cb, false);
+					modules.blocks.chain.applyBlock(block, false, cb);
 				}
 				// Update last block
 				modules.blocks.lastBlock.set(block);

modules/blocks/verify.js

@@ -7,6 +7,9 @@ var crypto = require('crypto');
 var slots = require('../../helpers/slots.js');
 var sql = require('../../sql/blocks.js');
 var exceptions = require('../../helpers/exceptions.js');
+var BSON = require('bson');
+
+var bson = new BSON();
 
 var modules, library, self, __private = {};
 
@@ -87,9 +90,47 @@ __private.checkTransaction = function (block, transaction, cb) {
 	});
 };
 
+/**
+ * Adds default properties to block.
+ * @private
+ * @param {Object} block Block object reduced
+ * @return {Object} Block object completed
+ */
+__private.addBlockProperties = function (block) {
+	if (block.version === undefined) {
+		block.version = 0;
+	}
+	if (block.numberOfTransactions === undefined) {
+		if (block.transactions === undefined) {
+			block.numberOfTransactions = 0;
+		} else {
+			block.numberOfTransactions = block.transactions.length;
+		}
+	}
+
+	return block;
+};
+
+/**
+ * Deletes default properties from block.
+ * @private
+ * @param {Object} block Block object completed
+ * @return {Object} Block object reduced
+ */
+__private.deleteBlockProperties = function (block) {
+	var reducedBlock = JSON.parse(JSON.stringify(block));
+	if (reducedBlock.version === 0) {
+		delete reducedBlock.version;
+	}
+	// verifyBlock ensures numberOfTransactions is transactions.length
+	if (reducedBlock.numberOfTransactions) {
+		delete reducedBlock.numberOfTransactions;
+	}
+	return reducedBlock;
+};
+
 /**
  * Verify block and return all possible errors related to block
- * 
  * @public
  * @method verifyBlock
  * @param  {Object}  block Full block
@@ -158,7 +199,7 @@ Verify.prototype.verifyBlock = function (block) {
 		totalAmount += transaction.amount;
 		totalFee += transaction.fee;
 	}
-
+	
 	if (payloadHash.digest().toString('hex') !== block.payloadHash) {
 		return 'Invalid payload hash';
 	}
@@ -227,6 +268,18 @@ Verify.prototype.processBlock = function (block, broadcast, cb, saveBlock) {
 	}
 
 	async.series({
+		addBlockProperties: function (seriesCb) {
+			if (!broadcast) {
+				try {
+					// set default properties
+					block = __private.addBlockProperties(block);
+				} catch (err) {
+					return setImmediate(seriesCb, err);
+				}
+			}
+
+			return setImmediate(seriesCb);
+		},
 		normalizeBlock: function (seriesCb) {
 			try {
 				block = library.logic.block.objectNormalize(block);
@@ -236,6 +289,20 @@ Verify.prototype.processBlock = function (block, broadcast, cb, saveBlock) {
 
 			return setImmediate(seriesCb);
 		},
+		deleteBlockProperties: function (seriesCb) {
+			if (broadcast) {
+				try {
+					// delete default properties
+					var blockReduced = __private.deleteBlockProperties(block);
+					var serializedBlockReduced = bson.serialize(blockReduced);
+					modules.blocks.chain.broadcastReducedBlock(serializedBlockReduced, broadcast);
+				} catch (err) {
+					return setImmediate(seriesCb, err);
+				}
+			}
+
+			return setImmediate(seriesCb);
+		},
 		verifyBlock: function (seriesCb) {
 			// Sanity check of the block, if values are coherent.
 			// No access to database
@@ -289,7 +356,7 @@ Verify.prototype.processBlock = function (block, broadcast, cb, saveBlock) {
 			// * Block and transactions have valid values (signatures, block slots, etc...)
 			// * The check against database state passed (for instance sender has enough LSK, votes are under 101, etc...)
 			// We thus update the database with the transactions values, save the block and tick it.
-			modules.blocks.chain.applyBlock(block, broadcast, cb, saveBlock);
+			modules.blocks.chain.applyBlock(block, saveBlock, cb);
 		}
 	});
 };

package.json

@@ -17,6 +17,7 @@
     "async": "=2.1.4",
     "bignumber.js": "=4.0.0",
     "body-parser": "=1.16.0",
+    "bson": "^1.0.4",
     "bytebuffer": "=5.0.1",
     "change-case": "=3.0.1",
     "colors": "=1.1.2",

test/common/initModule.js

@@ -28,13 +28,25 @@ var modulesLoader = new function () {
 		genesisblock: { block: genesisblock },
 		logger: this.logger,
 		network: {
-			app: express()
+			app: express(),
+			io: {
+				sockets: express()
+			}
 		},
 		public: '../../public',
 		schema: new z_schema(),
 		ed: ed,
 		bus: {
-			message: function () {}
+			argsMessages: [],
+			message: function () {
+				Array.prototype.push.apply(this.argsMessages, arguments);
+			},
+			getMessages: function () {
+				return this.argsMessages;
+			},
+			clearMessages: function () {
+				this.argsMessages = [];
+			}
 		},
 		nonce: randomString.generate(16)
 	};