OriginTrail v8.0.0-beta Release

ot-node.js

@@ -55,6 +55,7 @@ class OTNode {
 
         await this.initializeRouters();
         await this.startNetworkModule();
+        await this.initializeBLSService();
         this.startTelemetryModule();
         this.resumeCommandExecutor();
         this.logger.info('Node is up and running!');
@@ -403,6 +404,18 @@ class OTNode {
         tripleStoreService.initializeRepositories();
     }
 
+    async initializeBLSService() {
+        try {
+            const blsService = this.container.resolve('blsService');
+            await blsService.initialize();
+        } catch (error) {
+            this.logger.error(
+                `Unable to initialize BLS Service. Error message: ${error.message} OT-node shutting down...`,
+            );
+            this.stop(1);
+        }
+    }
+
     stop(code = 0) {
         this.logger.info('Stopping node...');
         process.exit(code);

package.json

@@ -80,7 +80,7 @@
         "axios": "^1.6.0",
         "cors": "^2.8.5",
         "deep-extend": "^0.6.0",
-        "dkg-evm-module": "^8.0.1-alpha.1",
+        "dkg-evm-module": "^8.0.2-alpha.0",
         "dotenv": "^16.0.1",
         "ethers": "^5.7.2",
         "express": "^4.18.1",

src/commands/cleaners/pending-storage-cleaner-command.js

@@ -0,0 +1,63 @@
+import Command from '../command.js';
+import {
+    PUBLISH_STORAGE_MEMORY_CLEANUP_COMMAND_CLEANUP_TIME_MILLS,
+    PUBLISH_STORAGE_FILE_CLEANUP_COMMAND_CLEANUP_TIME_MILLS,
+    PENDING_STORAGE_FILES_FOR_REMOVAL_MAX_NUMBER,
+} from '../../constants/constants.js';
+
+/**
+ * Cleans memory cache in the pending storage service
+ */
+class PendingStorageCleanerCommand extends Command {
+    constructor(ctx) {
+        super(ctx);
+        this.logger = ctx.logger;
+        this.pendingStorageService = ctx.pendingStorageService;
+    }
+
+    /**
+     * Executes command and produces one or more events
+     * @param command
+     */
+    async execute() {
+        this.logger.debug('Starting command for removal of expired pending storage entries');
+
+        const removed = await this.pendingStorageService.removeExpiredFileCache(
+            PUBLISH_STORAGE_FILE_CLEANUP_COMMAND_CLEANUP_TIME_MILLS,
+            PENDING_STORAGE_FILES_FOR_REMOVAL_MAX_NUMBER,
+        );
+        if (removed) {
+            this.logger.debug(`Successfully removed ${removed} expired cached operation files`);
+        }
+
+        return Command.repeat();
+    }
+
+    /**
+     * Recover system from failure
+     * @param command
+     * @param error
+     */
+    async recover(command) {
+        this.logger.warn(`Failed to clean pending storage: error: ${command.message}`);
+        return Command.repeat();
+    }
+
+    /**
+     * Builds default command
+     * @param map
+     * @returns {{add, data: *, delay: *, deadline: *}}
+     */
+    default(map) {
+        const command = {
+            name: 'pendingStorageCleanerCommand',
+            period: PUBLISH_STORAGE_MEMORY_CLEANUP_COMMAND_CLEANUP_TIME_MILLS,
+            data: {},
+            transactional: false,
+        };
+        Object.assign(command, map);
+        return command;
+    }
+}
+
+export default PendingStorageCleanerCommand;

src/commands/command-executor.js

@@ -7,6 +7,7 @@ import {
     COMMAND_STATUS,
     DEFAULT_COMMAND_DELAY_IN_MILLS,
     COMMAND_QUEUE_PARALLELISM,
+    DEFAULT_COMMAND_PRIORITY,
 } from '../constants/constants.js';
 
 /**
@@ -20,7 +21,7 @@ class CommandExecutor {
         this.repositoryModuleManager = ctx.repositoryModuleManager;
         this.verboseLoggingEnabled = ctx.config.commandExecutorVerboseLoggingEnabled;
 
-        this.queue = async.queue((command, callback = () => {}) => {
+        this.queue = async.priorityQueue((command, callback = () => {}) => {
             this._execute(command)
                 .then((result) => {
                     callback(result);
@@ -81,7 +82,7 @@ class CommandExecutor {
             commandId: command.id,
             commandName: command.name,
         };
-        if (command.data && command.data.operationId) {
+        if (command.data?.operationId !== undefined) {
             commandContext.operationId = command.data.operationId;
         }
         const loggerWithContext = this.logger.child(commandContext);
@@ -106,7 +107,7 @@ class CommandExecutor {
             });
             return;
         }
-        if (command.deadlineAt && now > command.deadlineAt) {
+        if (command.deadlineAt !== undefined && now > command.deadlineAt) {
             loggerWithContext.warn('Command is too late...');
             await this._update(command, {
                 status: COMMAND_STATUS.EXPIRED,
@@ -280,7 +281,7 @@ class CommandExecutor {
         let delay = addDelay ?? 0;
 
         if (delay > MAX_COMMAND_DELAY_IN_MILLS) {
-            if (command.readyAt == null) {
+            if (command.readyAt === undefined) {
                 command.readyAt = Date.now();
             }
             command.readyAt += delay;
@@ -290,16 +291,19 @@ class CommandExecutor {
         if (insert) {
             command = await this._insert(command);
         }
+
+        const commandPriority = command.priority ?? DEFAULT_COMMAND_PRIORITY;
+
         if (delay) {
             setTimeout(
                 (timeoutCommand) => {
-                    this.queue.push(timeoutCommand);
+                    this.queue.push(timeoutCommand, commandPriority);
                 },
                 delay,
                 command,
             );
         } else {
-            this.queue.push(command);
+            this.queue.push(command, commandPriority);
         }
     }
 
@@ -311,7 +315,7 @@ class CommandExecutor {
      */
     async _handleRetry(retryCommand, handler) {
         const command = retryCommand;
-        if (command.retries > 1) {
+        if (command.retries !== undefined && command.retries > 1) {
             command.data = handler.pack(command.data);
             await this._update(command, {
                 status: COMMAND_STATUS.PENDING,
@@ -337,7 +341,7 @@ class CommandExecutor {
      * @private
      */
     async _handleError(command, handler, error) {
-        if (command.retries > 0) {
+        if (command.retries !== undefined && command.retries > 0) {
             await this._update(command, {
                 retries: command.retries - 1,
             });
@@ -373,28 +377,26 @@ class CommandExecutor {
             [command.name] = command.sequence;
             command.sequence = command.sequence.slice(1);
         }
-        if (!command.readyAt) {
-            command.readyAt = Date.now(); // take current time
-        }
-        if (command.delay == null) {
-            command.delay = 0;
-        }
-        if (!command.transactional) {
-            command.transactional = 0;
-        }
+
+        command.readyAt = command.readyAt || Date.now();
+        command.delay = command.delay ?? 0;
+        command.transactional = command.transactional ?? 0;
+        command.priority = command.priority ?? DEFAULT_COMMAND_PRIORITY;
+        command.status = COMMAND_STATUS.PENDING;
+
         if (!command.data) {
             const commandInstance = this.commandResolver.resolve(command.name);
             if (commandInstance) {
                 command.data = commandInstance.pack(command.data);
             }
         }
-        command.status = COMMAND_STATUS.PENDING;
-        const opts = {};
-        if (transaction != null) {
-            opts.transaction = transaction;
-        }
+
+        const opts = transaction ? { transaction } : {};
+
         const model = await this.repositoryModuleManager.createCommand(command, opts);
+
         command.id = model.id;
+
         return command;
     }
 
@@ -462,6 +464,7 @@ class CommandExecutor {
                     id: commandModel.id,
                     name: commandModel.name,
                     data: commandModel.data,
+                    priority: commandModel.priority ?? DEFAULT_COMMAND_PRIORITY,
                     readyAt: commandModel.readyAt,
                     delay: commandModel.delay,
                     startedAt: commandModel.startedAt,

src/commands/local-store/local-store-command.js

@@ -1,4 +1,10 @@
-import { OPERATION_ID_STATUS, ERROR_TYPE, LOCAL_STORE_TYPES } from '../../constants/constants.js';
+import {
+    OPERATION_ID_STATUS,
+    ERROR_TYPE,
+    LOCAL_STORE_TYPES,
+    OPERATION_REQUEST_STATUS,
+    NETWORK_MESSAGE_TYPES,
+} from '../../constants/constants.js';
 import Command from '../command.js';
 
 class LocalStoreCommand extends Command {
@@ -8,6 +14,7 @@ class LocalStoreCommand extends Command {
         this.paranetService = ctx.paranetService;
         this.pendingStorageService = ctx.pendingStorageService;
         this.operationIdService = ctx.operationIdService;
+        this.operationService = ctx.publishService;
         this.dataService = ctx.dataService;
         this.ualService = ctx.ualService;
         this.serviceAgreementService = ctx.serviceAgreementService;
@@ -37,27 +44,34 @@ class LocalStoreCommand extends Command {
 
             if (storeType === LOCAL_STORE_TYPES.TRIPLE) {
                 const storePromises = [];
-                if (cachedData.public.dataset && cachedData.public.datasetRoot) {
-                    storePromises.push(
-                        this.pendingStorageService.cacheDataset(
-                            blockchain,
-                            cachedData.public.datasetRoot,
-                            cachedData.public.dataset,
-                            operationId,
-                        ),
-                    );
-                }
-                // if (cachedData.private?.assertion && cachedData.private?.assertionId) {
+
+                // if (cachedData.dataset && cachedData.datasetRoot) {
                 //     storePromises.push(
                 //         this.pendingStorageService.cacheDataset(
-                //             blockchain,
-                //             datasetRoot,
-                //             dataset,
                 //             operationId,
+                //             cachedData.datasetRoot,
+                //             cachedData.dataset,
                 //         ),
                 //     );
                 // }
+                // if (cachedData.private?.assertion && cachedData.private?.assertionId) {
+                //     storePromises.push(
+                //         this.pendingStorageService.cacheDataset(operationId, datasetRoot, dataset),
+                //     );
+                // }
                 await Promise.all(storePromises);
+                await this.operationService.processResponse(
+                    command,
+                    OPERATION_REQUEST_STATUS.COMPLETED,
+                    {
+                        messageType: NETWORK_MESSAGE_TYPES.RESPONSES.ACK,
+                        messageData: {
+                            signature: `signature-${Math.floor(Math.random() * 1000000) + 1}`,
+                        },
+                    },
+                    null,
+                    true,
+                );
             } else if (storeType === LOCAL_STORE_TYPES.TRIPLE_PARANET) {
                 const paranetMetadata = await this.blockchainModuleManager.getParanetMetadata(
                     blockchain,
@@ -73,7 +87,7 @@ class LocalStoreCommand extends Command {
                 await this.tripleStoreModuleManager.initializeParanetRepository(paranetRepository);
                 await this.paranetService.initializeParanetRecord(blockchain, paranetId);
 
-                if (cachedData.public.dataset && cachedData.public.datasetRoot) {
+                if (cachedData && cachedData.datasetRoot) {
                     // await this.tripleStoreService.localStoreAsset(
                     //     paranetRepository,
                     //     cachedData.public.assertionId,
@@ -86,7 +100,7 @@ class LocalStoreCommand extends Command {
                     //     LOCAL_INSERT_FOR_CURATED_PARANET_RETRY_DELAY,
                     // );
                 }
-                if (cachedData.private?.assertion && cachedData.private?.assertionId) {
+                if (cachedData && cachedData.datasetRoot) {
                     // await this.tripleStoreService.localStoreAsset(
                     //     paranetRepository,
                     //     cachedData.private.assertionId,