Merge pull request #1801 from OriginTrail/v6/prerelease/testnet

OriginTrail 6.0.0-beta.1.28 Testnet Release

config/config.json

@@ -24,7 +24,7 @@
       "password": ""
     },
     "logLevel": "trace",
-    "replicationFactor" : 3,
+    "replicationFactor" : 5,
     "rpcPort": 8900,
     "network": {
       "port": 9000,
@@ -65,7 +65,7 @@
       "password": ""
     },
     "logLevel": "trace",
-    "replicationFactor" : 3,
+    "replicationFactor" : 5,
     "rpcPort": 8900,
     "network": {
       "port": 9000,
@@ -101,7 +101,7 @@
       "password": ""
     },
     "logLevel": "trace",
-    "replicationFactor" : 3,
+    "replicationFactor" : 5,
     "rpcPort": 8900,
     "network": {
       "port": 9000,

external/blazegraph-service.js

@@ -31,7 +31,10 @@ class BlazegraphService {
 
             await axios(this.config.axios).then((response) => true)
                 .catch((error) => {
-                    this.logger.error(`Failed to write into Blazegraph: ${error} - ${error.stack}`);
+                    this.logger.error({
+                        msg: `Failed to write into Blazegraph: ${error} - ${error.stack}`,
+                        Event_name: constants.ERROR_TYPE.TRIPLE_STORE_INSERT_ERROR,
+                    });
                     return false;
                 });
         }

external/libp2p-service.js

@@ -147,7 +147,7 @@ class Libp2pService {
     async handleMessage(eventName, handler, options) {
         this.logger.info(`Enabling network protocol: ${eventName}`);
 
-        let async = false, timeout = 5e3;
+        let async = false, timeout = 60e3;
         if (options) {
             async = options.async;
             timeout = options.timeout;
@@ -193,8 +193,9 @@ class Libp2pService {
                     }
                 }
             } catch (e) {
+                const stringifiedData = await this.workerPool.exec('JSONStringify', [data]);
                 this.logger.error({
-                   msg: `Error: ${e}, stack: ${e.stack} \n Data received: ${data}`,
+                   msg: `Error: ${e}, stack: ${e.stack} \n Data received: ${stringifiedData}`,
                    Event_name: constants.ERROR_TYPE.LIBP2P_HANDLE_MSG_ERROR,
                 });
                 await pipe(

modules/command/common/keep-alive-command.js

@@ -1,10 +1,18 @@
 const { v1: uuidv1 } = require('uuid');
+const PeerId = require('peer-id');
+var axios = require('axios');
+const fs = require('fs');
 const Command = require('../command');
+const pjson = require('../../../package.json');
+const Models = require('../../../models/index');
+const constants = require('../../constants');
 
 class KeepAliveCommand extends Command {
     constructor(ctx) {
         super(ctx);
         this.logger = ctx.logger;
+        this.config = ctx.config;
+        this.validationService = ctx.validationService;
     }
 
     /**
@@ -19,6 +27,55 @@ class KeepAliveCommand extends Command {
             msg: message, Event_name: 'keep_alive', Operation_name: 'KeepAlive', Id_operation,
         });
 
+        const signalingMessage = {
+            nodeVersion: pjson.version,
+            autoUpdate: { enabled: this.config.autoUpdate.enabled },
+            telemetry: {
+                enabled: this.config.telemetryHub.enabled,
+            },
+        };
+        try {
+            if (!this.config.network.privateKey) {
+                const configFile = JSON.parse(fs.readFileSync(this.config.config ? this.config.config : '.origintrail_noderc'));
+                this.config.network.privateKey = configFile.network.privateKey;
+            }
+            const peerId = await PeerId.createFromPrivKey(this.config.network.privateKey);
+            signalingMessage.issuerWallet = this.config.blockchain[0].publicKey;
+            signalingMessage.kademliaNodeId = peerId._idB58String;
+            signalingMessage.nodeVersion = pjson.version;
+            signalingMessage.telemetry.latestAssertions = (await Models.assertions.findAll({
+                limit: 5,
+                order: [
+                    ['created_at', 'DESC'],
+                ],
+                attributes: ['hash', 'topics', 'created_at'],
+            })).map(x => ({assertionId: x.dataValues.hash, keyword: x.dataValues.topics, publishTimestamp: x.dataValues.created_at}));
+        } catch (e) {
+            this.logger.error({
+                msg: `An error has occurred with signaling data error: ${e}, stack: ${e.stack}`,
+                Event_name: constants.ERROR_TYPE.KEEP_ALIVE_ERROR,
+            });
+        }
+
+        const proof = {};
+        proof.hash = this.validationService.calculateHash(signalingMessage);
+        proof.signature = this.validationService.sign(proof.hash);
+
+        signalingMessage.proof = proof;
+
+        const config = {
+            method: 'post',
+            url: 'https://signum.origintrail.io:3000/signal',
+            headers: {
+                'Content-Type': 'application/json'
+            },
+            data: JSON.stringify(signalingMessage)
+        };
+
+        const that = this;
+        axios(config).catch(e=>{
+            that.handleError(uuidv1(), e, constants.ERROR_TYPE.KEEP_ALIVE_ERROR, false)
+        });
         return Command.repeat();
     }
 
@@ -34,7 +91,7 @@ class KeepAliveCommand extends Command {
             data: {
                 message: 'OT-Node is alive...',
             },
-            period: 1 * 60 * 1000,
+            period: 15 * 60 * 1000,
             transactional: false,
         };
         Object.assign(command, map);

modules/command/publish/insert-assertion-command.js

@@ -15,14 +15,21 @@ class InsertAssertionCommand extends Command {
      * @param command
      */
     async execute(command) {
-        const { documentPath, handlerId } = command.data;
+        const { documentPath, handlerId, operationId } = command.data;
         let { nquads, assertion } = await this.fileService.loadJsonFromFile(documentPath);
 
         try {
             await this.dataService.insert(nquads.join('\n'), `${constants.DID_PREFIX}:${assertion.id}`);
             this.logger.info(`Assertion ${assertion.id} has been successfully inserted`);
         } catch (e) {
             await this.handleError(handlerId, e, constants.ERROR_TYPE.INSERT_ASSERTION_ERROR, true);
+            this.logger.emit({
+                msg: 'Finished measuring execution of publish command',
+                Event_name: 'publish_end',
+                Operation_name: 'publish',
+                Id_operation: operationId,
+            });
+            return Command.empty();
         }
 
         return this.continueSequence(command.data, command.sequence);

modules/command/publish/send-assertion-command.js

@@ -18,7 +18,7 @@ class SendAssertionCommand extends Command {
      * @param command
      */
     async execute(command) {
-        const { documentPath, handlerId } = command.data;
+        const { documentPath, handlerId, operationId } = command.data;
 
         let { nquads, assertion } = await this.fileService.loadJsonFromFile(documentPath);
 
@@ -45,7 +45,7 @@ class SendAssertionCommand extends Command {
         nodes = [...new Set(nodes)];
 
         for (const node of nodes) {
-            this.publishService.store({ id:assertion.id, nquads: nquads }, node).catch((e) => {
+            this.publishService.store({ id: assertion.id, nquads: nquads }, node).catch((e) => {
                 this.handleError(handlerId, e, `Error while sending data with assertion id ${assertion.id} to node ${node._idB58String}. Error message: ${e.message}. ${e.stack}`);
             });
         }
@@ -59,6 +59,20 @@ class SendAssertionCommand extends Command {
                 },
             },
         );
+        this.logger.emit({
+            msg: 'Finished measuring execution of publish command',
+            Event_name: 'publish_end',
+            Operation_name: 'publish',
+            Id_operation: operationId,
+        });
+
+        if (command.data.isTelemetry) {
+            await Models.assertions.create({
+                hash: assertion.id,
+                topics: JSON.stringify(assertion.metadata.keywords[0]),
+                createdAt: assertion.metadata.timestamp,
+            });
+        }
 
         return this.continueSequence(command.data, command.sequence);
     }

modules/command/publish/submit-proofs-command.js

@@ -1,7 +1,7 @@
+const sortedStringify = require('json-stable-stringify');
 const Command = require('../command');
 const Models = require('../../../models/index');
 const constants = require('../../constants');
-const sortedStringify = require("json-stable-stringify");
 
 class SubmitProofsCommand extends Command {
     constructor(ctx) {
@@ -20,21 +20,22 @@ class SubmitProofsCommand extends Command {
      * @param command
      */
     async execute(command) {
-        const { documentPath, handlerId, method, isUrgent } = command.data;
+        const {
+            documentPath, handlerId, method, isTelemetry, operationId,
+        } = command.data;
 
         try {
             let { nquads, assertion } = await this.fileService.loadJsonFromFile(documentPath);
 
-
-            this.logger.info(`Sending transaction to the blockchain`);
+            this.logger.info('Sending transaction to the blockchain');
             let result;
-            if (isUrgent){
-                result = await this.blockchainQueue.unshift({method, assertion});
-            }else {
-                if (this.blockchainQueue.length()>constants.BLOCKCHAIN_QUEUE_LIMIT){
-                    throw new Error ('Blockchain queue is full');
+            if (isTelemetry) {
+                result = await this.blockchainQueue.unshift({ method, assertion });
+            } else {
+                if (this.blockchainQueue.length() > constants.BLOCKCHAIN_QUEUE_LIMIT) {
+                    throw new Error('Blockchain queue is full');
                 }
-                result = await this.blockchainQueue.push({method, assertion});
+                result = await this.blockchainQueue.push({ method, assertion });
             }
 
             const { transactionHash, blockchain } = result;
@@ -50,20 +51,24 @@ class SubmitProofsCommand extends Command {
 
             await this.fileService
                 .writeContentsToFile(handlerIdCachePath, handlerId, sortedStringify({
-                    nquads, assertion
+                    nquads, assertion,
                 }));
-
         } catch (e) {
             await this.handleError(handlerId, e, constants.ERROR_TYPE.SUBMIT_PROOFS_ERROR, true);
-
+            this.logger.emit({
+                msg: 'Finished measuring execution of publish command',
+                Event_name: 'publish_end',
+                Operation_name: 'publish',
+                Id_operation: operationId,
+            });
             return Command.empty();
         }
 
         return this.continueSequence(command.data, command.sequence);
     }
 
-    async sendTransaction (args){
-        const { assertion, method} = args;
+    async sendTransaction(args) {
+        const { assertion, method } = args;
         let result;
         switch (method) {
             case 'publish':

modules/constants.js

@@ -63,7 +63,6 @@ exports.TRIPLE_STORE_CONNECT_MAX_RETRIES = 10;
  */
 exports.TRIPLE_STORE_CONNECT_RETRY_FREQUENCY = 10; // 10 seconds
 
-
 /**
  * @constant {number} BLOCKCHAIN_QUEUE_LIMIT
  * - Blockchain queue limit
@@ -101,6 +100,7 @@ exports.ERROR_TYPE = {
     NODE_INFO_ROUTE_ERROR: 'NodeInfoRouteError',
     EXTRACT_METADATA_ERROR: 'ExtractMetadataError',
     TRIPLE_STORE_UNAVAILABLE_ERROR: 'TripleStoreUnavailableError',
+    TRIPLE_STORE_INSERT_ERROR: 'TripleStoreInsertError',
     LIBP2P_HANDLE_MSG_ERROR: 'Libp2pHandleMessageError',
     VERIFY_ASSERTION_ERROR: 'VerifyAssertionError',
     BLOCKCHAIN_CHECK_ERROR: 'BlockchainCheckError',
@@ -113,4 +113,5 @@ exports.ERROR_TYPE = {
     BLOCKCHAIN_INITIALIZATION_ERROR: 'BlockchainInitializationError',
     COMMAND_EXECUTOR_INITIALIZATION_ERROR: 'CommandExecutorInitializationError',
     RPC_INITIALIZATION_ERROR: 'RpcInitializationError',
+    KEEP_ALIVE_ERROR: 'KeepAliveError',
 };