Merge pull request #2854 from OriginTrail/v6/prerelease/mainnet

OriginTrail Mainnet Release v6.1.1

ot-node.js

@@ -54,6 +54,12 @@ class OTNode {
         await this.initializeCommandExecutor();
         await this.initializeShardingTableService();
 
+        await MigrationExecutor.executeMarkStakingEventsAsProcessedMigration(
+            this.container,
+            this.logger,
+            this.config,
+        );
+
         MigrationExecutor.executeUalExtensionTripleStoreMigration(
             this.container,
             this.logger,
@@ -62,6 +68,12 @@ class OTNode {
             await this.initializeBlockchainEventListenerService();
         });
 
+        await MigrationExecutor.executePullShardingTableMigration(
+            this.container,
+            this.logger,
+            this.config,
+        );
+
         await this.initializeRouters();
         await this.startNetworkModule();
         this.startTelemetryModule();

package.json

@@ -1,6 +1,6 @@
 {
     "name": "origintrail_node",
-    "version": "6.1.0",
+    "version": "6.1.1",
     "description": "OTNode V6",
     "main": "index.js",
     "type": "module",

src/migration/mark-staking-events-as-processed-migration.js

@@ -0,0 +1,25 @@
+/* eslint-disable no-await-in-loop */
+import BaseMigration from './base-migration.js';
+
+class MarkStakingEventsAsProcessedMigration extends BaseMigration {
+    constructor(migrationName, logger, config, repositoryModuleManager, blockchainModuleManager) {
+        super(migrationName, logger, config);
+        this.repositoryModuleManager = repositoryModuleManager;
+        this.blockchainModuleManager = blockchainModuleManager;
+    }
+
+    async executeMigration() {
+        this.logger.info('Marking old blockchain events as processed');
+        for (const blockchain of this.blockchainModuleManager.getImplementationNames()) {
+            const timestamp = Date.now();
+            const block = await this.blockchainModuleManager.getLatestBlock(blockchain);
+            const query = `update blockchain
+                           set last_checked_block     = ${block.number},
+                               last_checked_timestamp = ${timestamp}
+                           where blockchain_id = 'otp:2043'`;
+            await this.repositoryModuleManager.query(query);
+        }
+    }
+}
+
+export default MarkStakingEventsAsProcessedMigration;

src/migration/migration-executor.js

@@ -14,6 +14,7 @@ import ServiceAgreementsDataInspector from './service-agreements-data-inspector.
 import ServiceAgreementsInvalidDataMigration from './service-agreements-invalid-data-migration.js';
 import UalExtensionUserConfigurationMigration from './ual-extension-user-configuration-migration.js';
 import UalExtensionTripleStoreMigration from './ual-extension-triple-store-migration.js';
+import MarkStakingEventsAsProcessedMigration from './mark-staking-events-as-processed-migration.js';
 
 class MigrationExecutor {
     static async executePullShardingTableMigration(container, logger, config) {
@@ -28,7 +29,7 @@ class MigrationExecutor {
         const validationModuleManager = container.resolve('validationModuleManager');
 
         const migration = new PullBlockchainShardingTableMigration(
-            'pullShardingTableMigrationV613',
+            'pullShardingTableMigrationV611',
             logger,
             config,
             repositoryModuleManager,
@@ -338,6 +339,31 @@ class MigrationExecutor {
         }
     }
 
+    static async executeMarkStakingEventsAsProcessedMigration(container, logger, config) {
+        if (process.env.NODE_ENV !== NODE_ENVIRONMENTS.MAINNET) return;
+
+        const repositoryModuleManager = container.resolve('repositoryModuleManager');
+        const blockchainModuleManager = container.resolve('blockchainModuleManager');
+
+        const migration = new MarkStakingEventsAsProcessedMigration(
+            'markStakingEventsAsProcessedMigration',
+            logger,
+            config,
+            repositoryModuleManager,
+            blockchainModuleManager,
+        );
+        if (!(await migration.migrationAlreadyExecuted())) {
+            try {
+                await migration.migrate();
+            } catch (error) {
+                logger.error(
+                    `Unable to execute mark staking events as processed migration. Error: ${error.message}`,
+                );
+                this.exitNode(1);
+            }
+        }
+    }
+
     static exitNode(code = 0) {
         process.exit(code);
     }

src/modules/repository/implementation/sequelize/repositories/blockchain-event-repository.js

@@ -7,20 +7,30 @@ class BlockchainEventRepository {
     }
 
     async insertBlockchainEvents(events) {
-        const inserted = await this.model.bulkCreate(
-            events.map((event) => ({
-                contract: event.contract,
-                event: event.event,
-                data: event.data,
-                block: event.block,
-                blockchainId: event.blockchainId,
-                processed: false,
-            })),
-            {
-                ignoreDuplicates: true,
-            },
-        );
-        return inserted.map((event) => event.dataValues);
+        const chunkSize = 10000;
+        let insertedEvents = [];
+
+        for (let i = 0; i < events.length; i += chunkSize) {
+            const chunk = events.slice(i, i + chunkSize);
+            // eslint-disable-next-line no-await-in-loop
+            const insertedChunk = await this.model.bulkCreate(
+                chunk.map((event) => ({
+                    contract: event.contract,
+                    event: event.event,
+                    data: event.data,
+                    block: event.block,
+                    blockchainId: event.blockchainId,
+                    processed: false,
+                })),
+                {
+                    ignoreDuplicates: true,
+                },
+            );
+
+            insertedEvents = insertedEvents.concat(insertedChunk.map((event) => event.dataValues));
+        }
+
+        return insertedEvents;
     }
 
     async getAllUnprocessedBlockchainEvents(eventNames, blockchainId) {