Merge pull request #270 from lidofinance/feat/improve-tasks-processing

Feat/improve tasks processing

package.json

@@ -1,6 +1,6 @@
 {
   "name": "lido-council-daemon",
-  "version": "3.2.0",
+  "version": "3.3.0",
   "description": "Lido Council Daemon",
   "author": "Lido team",
   "private": true,

src/contracts/data-bus/data-bus.service.ts

@@ -53,7 +53,7 @@ export class DataBusService {
 
     const dataBusClient = new DataBusClient(this.dataBusAddress, this.wallet);
     this.dsmMessageSender = new DSMMessageSender(dataBusClient);
-    await this.monitorGuardianBalance();
+    await this.monitorGuardianDataBusBalance();
     this.subscribeToEVMChainUpdates();
   }
 
@@ -64,7 +64,7 @@ export class DataBusService {
     const provider = this.provider;
     provider.on('block', async (blockNumber) => {
       if (blockNumber % DATA_BUS_BALANCE_UPDATE_BLOCK_RATE !== 0) return;
-      await this.monitorGuardianBalance().catch((error) =>
+      await this.monitorGuardianDataBusBalance().catch((error) =>
         this.logger.error(error),
       );
     });
@@ -77,7 +77,7 @@ export class DataBusService {
    * Updates the account balance metric.
    */
   @OneAtTime()
-  public async monitorGuardianBalance() {
+  public async monitorGuardianDataBusBalance() {
     const balanceWei = await this.getAccountBalance();
     const balanceETH = formatEther(balanceWei);
     const { chainId } = await this.provider.getNetwork();

src/guardian/guardian.constants.ts

@@ -1,7 +1,5 @@
-import { CronExpression } from '@nestjs/schedule';
-
 export const GUARDIAN_DEPOSIT_RESIGNING_BLOCKS = 10;
 export const GUARDIAN_DEPOSIT_JOB_NAME = 'guardian-deposit-job';
-export const GUARDIAN_DEPOSIT_JOB_DURATION = CronExpression.EVERY_5_SECONDS;
+export const GUARDIAN_DEPOSIT_JOB_DURATION_MS = 5000;
 export const MIN_KAPI_VERSION = '2.2.0';
 export const GUARDIAN_PING_BLOCKS_PERIOD = 300;

src/guardian/guardian.service.ts

@@ -7,12 +7,11 @@ import {
 import { compare } from 'compare-versions';
 import { WINSTON_MODULE_NEST_PROVIDER } from 'nest-winston';
 import { SchedulerRegistry } from '@nestjs/schedule';
-import { CronJob } from 'cron';
 import { DepositRegistryService } from 'contracts/deposits-registry';
 import { SecurityService } from 'contracts/security';
 import { RepositoryService } from 'contracts/repository';
 import {
-  GUARDIAN_DEPOSIT_JOB_DURATION,
+  GUARDIAN_DEPOSIT_JOB_DURATION_MS,
   GUARDIAN_DEPOSIT_JOB_NAME,
 } from './guardian.constants';
 import { OneAtTime } from 'common/decorators';
@@ -127,25 +126,22 @@ export class GuardianService implements OnModuleInit {
    * Subscribes to the staking router modules updates
    */
   public subscribeToModulesUpdates() {
-    const cron = new CronJob(GUARDIAN_DEPOSIT_JOB_DURATION, () => {
-      this.handleNewBlock().catch((error) => {
-        this.logger.error(error);
-      });
-    });
-
-    this.logger.log('GuardianService subscribed to Ethereum events');
+    const interval = setInterval(
+      () => this.handleNewBlock().catch((error) => this.logger.error(error)),
+      GUARDIAN_DEPOSIT_JOB_DURATION_MS,
+    );
 
-    cron.start();
+    this.schedulerRegistry.addInterval(GUARDIAN_DEPOSIT_JOB_NAME, interval);
 
-    this.schedulerRegistry.addCronJob(GUARDIAN_DEPOSIT_JOB_NAME, cron);
+    this.logger.log('GuardianService subscribed to Ethereum events');
   }
 
   /**
    * Handles the appearance of a new block in the network
    */
   @OneAtTime()
   public async handleNewBlock(): Promise<void> {
-    this.logger.log('New staking router state cycle start');
+    this.logger.log('Beginning of the processing of the new Guardian cycle');
 
     try {
       const endTimer = this.jobDurationMetric
@@ -216,10 +212,15 @@ export class GuardianService implements OnModuleInit {
       // run key checks and send deposit messages to the queue without waiting.
       this.handleKeys(stakingModulesData, blockData, lidoKeys)
         .catch(this.logger.error)
-        .finally(() => endTimer());
+        .finally(() => {
+          this.logger.log('End of unvetting and deposits processing by Guardian');
+          endTimer()
+        });
     } catch (error) {
-      this.logger.error('Staking router state update error');
+      this.logger.error('Guardian cycle processing error');
       this.logger.error(error);
+    } finally {
+      this.logger.log('End of pause processing by Guardian');
     }
   }
 
@@ -288,8 +289,6 @@ export class GuardianService implements OnModuleInit {
       blockHash,
       blockNumber,
     });
-
-    this.logger.log('New staking router state cycle end');
   }
 
   private async checkKeys(