queue signature validation per-channel to ensure correct ordering

lib/historyKeeper.js

@@ -38,6 +38,7 @@ module.exports.create = function (config, cb) {
         channel_cache: {},
         queueStorage: WriteQueue(),
         queueDeletes: WriteQueue(),
+        queueValidation: WriteQueue(),
 
         batchIndexReads: BatchRead("HK_GET_INDEX"),
         batchMetadata: BatchRead('GET_METADATA'),

lib/hk-util.js

@@ -859,15 +859,21 @@ HK.onChannelMessage = function (Env, Server, channel, msgStruct) {
 
         // trim the checkpoint indicator off the message if it's present
         let signedMsg = (isCp) ? msgStruct[4].replace(CHECKPOINT_PATTERN, '') : msgStruct[4];
-        // convert the message from a base64 string into a Uint8Array
-
-        //const txid = Util.uid();
 
         // Listen for messages
-        //console.log(+new Date(), "Send verification request");
-        Env.validateMessage(signedMsg, metadata.validateKey, w(function (err) {
+        /*  queueing this helps avoid race conditions in which workers
+            validate and write messages in a different order than they were received.
+            For best effect the validate and store should actually be queued atomically,
+            but this is a step in the right direction.
+        */
+        var proceed = w();
+        Env.queueValidation(channel.id, function (next) {
+            Env.validateMessage(signedMsg, metadata.validateKey, function (err) {
+                // always go on to the next item in the queue regardless of the outcome
+                next();
+
                 // no errors means success
-            if (!err) { return; }
+                if (!err) { return proceed(); }
                 // validation can fail in multiple ways
                 if (err === 'FAILED') {
                     // we log this case, but not others for some reason
@@ -875,7 +881,8 @@ HK.onChannelMessage = function (Env, Server, channel, msgStruct) {
                 }
                 // always abort if there was an error...
                 return void w.abort();
-        }));
+            });
+        });
     }).nThen(function () {
         // do checkpoint stuff...