heartbeat + enhanced tests + refactor

bin/validatorWorker.js

@@ -67,7 +67,6 @@ function validatorTick(channel) {
 	const tick = isLeader ? leader.tick : follower.tick
 	return tick(adapter, channel)
 }
-
 function wait(ms) {
 	return new Promise((resolve, _) => setTimeout(resolve, ms))
 }

routes/channel.js

@@ -156,11 +156,14 @@ function isValidatorMsgValid(msg) {
 	// @TODO either make this more sophisticated, or rewrite this in a type-safe lang
 	// for example, we should validate if every value in balances is a positive integer
 	return msg
-		&& typeof(msg.stateRoot) === 'string' && msg.stateRoot.length == 64
+		&& (
+			(typeof(msg.stateRoot) === 'string' && msg.stateRoot.length == 64)
+			|| typeof(msg.timestamp) === 'string' 
+		)
 		&& typeof(msg.signature) === 'string'
 		&& (
 			(msg.type === 'NewState' && typeof(msg.balances) === 'object')
-			|| msg.type === 'ApproveState'
+			|| msg.type === 'ApproveState' || msg.type === 'HeartBeat'
 		)
 }
 

services/validatorWorker/follower.js

@@ -1,9 +1,10 @@
-const BN = require('bn.js')
 const assert = require('assert')
 const db = require('../../db')
 const { persistAndPropagate } = require('./lib/propagation')
-const { isValidTransition, isHealthy, isValidRootHash } = require('./lib/followerRules')
+const { isValidTransition, isHealthy } = require('./lib/followerRules')
+const { isValidRootHash, toBNMap } = require('./lib')
 const producer = require('./producer')
+const heartbeat = require('./heartbeat')
 
 function tick(adapter, channel) {
 	// @TODO: there's a flaw if we use this in a more-than-two validator setup
@@ -28,6 +29,15 @@ function tick(adapter, channel) {
 			return onNewState(adapter, { ...res, newMsg, approveMsg })
 		})
 	})
+	.then(function(res){
+		if(res && res.nothingNew){
+			// send heartbeat
+			return heartbeat(adapter, channel)
+			.then(() => res)
+		} else {
+			return res
+		}
+	})
 }
 
 function onNewState(adapter, {channel, balances, newMsg, approveMsg}) {
@@ -57,7 +67,7 @@ function onNewState(adapter, {channel, balances, newMsg, approveMsg}) {
 			console.error(`validatatorWorker: ${channel.id}: invalid signature NewState`, prevBalances, newBalances)
 			return { nothingNew: true }
 		}
-
+    
 		const stateRootRaw = Buffer.from(stateRoot, 'hex')
 		return adapter.sign(stateRootRaw)
 		.then(function(signature) {
@@ -71,13 +81,6 @@ function onNewState(adapter, {channel, balances, newMsg, approveMsg}) {
 	})
 }
 
-function toBNMap(raw) {
-	assert.ok(raw && typeof(raw) === 'object', 'raw map is a valid object')
-	const balances = {}
-	Object.entries(raw).forEach(([acc, bal]) => balances[acc] = new BN(bal, 10))
-	return balances
-}
-
 // @TODO getLatestMsg should be a part of a DB abstraction so we can use it in other places too
 // e.g. validating on POST /validator-messages (to get the previous), and a public API to get the latest msgs of a type
 function getLatestMsg(channelId, from, type) {

services/validatorWorker/heartbeat.js

@@ -0,0 +1,35 @@
+const assert = require('assert')
+const { persistAndPropagate } = require('./lib/propagation')
+
+function heartbeat(adapter, channel){
+	const whoami = adapter.whoami()
+	const validatorIdx = channel.validators.indexOf(whoami)
+	assert.ok(validatorIdx !== -1, 'validatorTick: sending heartbeat for a channel where we are not validating')
+	const otherValidators = channel.spec.validators.filter(v => v.id != whoami)
+
+	let timestamp = Buffer.alloc(32);
+	timestamp.writeUIntBE(Date.now(), 26, 6);
+
+	// in the future, we can add more information to this tree, 
+	// such as the validator node capacity/status, 
+	// or a proof of 'no earlier than' (hash of the latest blockchain block)
+	const tree = new adapter.MerkleTree([ timestamp ])
+	const infoRootRaw = tree.getRoot()
+
+	const stateRootRaw = adapter.getSignableStateRoot(Buffer.from(channel.id), infoRootRaw)
+
+	return adapter.sign(stateRootRaw)
+	.then(function(signature) {
+		const stateRoot = stateRootRaw.toString('hex')
+		timestamp = timestamp.toString('hex')
+
+		return persistAndPropagate(adapter, otherValidators, channel, {
+			type: 'HeartBeat',
+			timestamp,
+			signature,
+			stateRoot,
+		});
+	})
+}
+
+module.exports = heartbeat

services/validatorWorker/leader.js

@@ -1,14 +1,24 @@
 const { persistAndPropagate } = require('./lib/propagation')
-const { getStateRootHash } = require('./lib/followerRules')
-
+const { getStateRootHash } = require('./lib')
 const producer = require('./producer')
+const heartbeat = require('./heartbeat')
 
 function tick(adapter, channel) {
 	return producer.tick(channel)
 		.then(
 			res => res.newStateTree ?
 				afterProducer(adapter, res)
 				: { nothingNew: true }
+		).then(
+			res => {
+				if(res && res.nothingNew){
+					// send heartbeat
+					return heartbeat(adapter, channel)
+					.then(() => res)
+				} else {
+					return res
+				}
+			}
 		)
 }
 

services/validatorWorker/lib/followerRules.js

@@ -47,21 +47,6 @@ function sumMins(our, approved) {
 	)
 }
 
-function getStateRootHash(channel, balances, adapter){
-	// Note: MerkleTree takes care of deduplicating and sorting
-	const elems = Object.keys(balances).map(
-		acc => adapter.getBalanceLeaf(acc, balances[acc])
-	)
-	const tree = new adapter.MerkleTree(elems)
-	const balanceRoot = tree.getRoot()
-	// keccak256(channelId, balanceRoot)
-	const stateRoot = adapter.getSignableStateRoot(Buffer.from(channel.id), balanceRoot).toString('hex')
-	return stateRoot
-}
-
-function isValidRootHash(leaderRootHash, { channel, balances, adapter }) {
-	return getStateRootHash(channel, balances, adapter) === leaderRootHash
-}
 
 
-module.exports = { isValidTransition, isHealthy, isValidRootHash, getStateRootHash }
+module.exports = { isValidTransition, isHealthy }

services/validatorWorker/lib/index.js

@@ -0,0 +1,28 @@
+const assert = require('assert')
+const BN = require('bn.js')
+
+function getStateRootHash(channel, balances, adapter){
+	// Note: MerkleTree takes care of deduplicating and sorting
+	const elems = Object.keys(balances).map(
+		acc => adapter.getBalanceLeaf(acc, balances[acc])
+	)
+	const tree = new adapter.MerkleTree(elems)
+	const balanceRoot = tree.getRoot()
+	// keccak256(channelId, balanceRoot)
+	const stateRoot = adapter.getSignableStateRoot(Buffer.from(channel.id), balanceRoot).toString('hex')
+	return stateRoot
+}
+
+function isValidRootHash(leaderRootHash, { channel, balances, adapter }) {
+	return getStateRootHash(channel, balances, adapter) === leaderRootHash
+}
+
+function toBNMap(raw) {
+	assert.ok(raw && typeof(raw) === 'object', 'raw map is a valid object')
+	const balances = {}
+	Object.entries(raw).forEach(([acc, bal]) => balances[acc] = new BN(bal, 10))
+	return balances
+}
+
+
+module.exports = { getStateRootHash, isValidRootHash, toBNMap }

test/index.js

@@ -3,7 +3,8 @@ const tape = require('tape')
 
 const BN = require('bn.js')
 const { isValidTransition, isHealthy } = require('../services/validatorWorker/lib/followerRules')
-
+const { getStateRootHash } = require('../services/validatorWorker/lib')
+const dummyAdapter = require('../adapters/dummy')
 const channel = { depositAmount: new BN(100) }
 
 tape('isValidTransition: empty to empty', function(t) {
@@ -80,5 +81,37 @@ tape('isHealthy: they have the same sum, but different entities are earning', fu
 	t.end()
 })
 
+//
+// State Root Hash
+//
+tape('getStateRootHash: returns correct result', function(t) {
+	[
+		{
+			channel: {
+				id: "testing"
+			},
+			balances: {
+				"publisher": 1,
+				"tester": 2
+			},
+			expectedHash: "da9b42bb60da9622404cade0aec4cda0a10104c6ec5f07ad67de081abb58c803"
+		},
+		{
+			channel: {
+				id: "fake"
+			},
+			balances: {
+				"publisher": 0,
+			},
+			expectedHash: "0b64767e909e9f36ab9574e6b93921390c40a0d899c3587db3b2df077b8e87d7"
+		}
+	].forEach(({ expectedHash, channel, balances }) => {
+		const actualHash = getStateRootHash(channel, balances, dummyAdapter)
+		t.equal(actualHash, expectedHash, "correct root hash")
+	});
+
+	t.end()
+})
+
 // @TODO: event aggregator
 // @TODO: producer, possibly leader/follower; mergePayableIntoBalances