feat: guardian module refactoring

src/common/prometheus/prometheus.provider.ts

@@ -66,25 +66,25 @@ export const PrometheusBuildInfoGaugeProvider = makeCounterProvider({
 export const PrometheusValidatedDepositsProvider = makeGaugeProvider({
   name: METRIC_VALIDATED_DEPOSITS_TOTAL,
   help: 'Number of deposits by validation',
-  labelNames: ['type'] as const,
+  labelNames: ['type', 'stakingModuleId'] as const,
 });
 
 export const PrometheusIntersectionsProvider = makeGaugeProvider({
   name: METRIC_INTERSECTIONS_TOTAL,
   help: 'Number of keys intersections',
-  labelNames: ['type'] as const,
+  labelNames: ['type', 'stakingModuleId'] as const,
 });
 
 export const PrometheusDepositedKeysProvider = makeGaugeProvider({
   name: METRIC_DEPOSITED_KEYS_TOTAL,
   help: 'Number of keys in the deposit contract',
-  labelNames: ['type'] as const,
+  labelNames: ['type', 'stakingModuleId'] as const,
 });
 
 export const PrometheusOperatorsKeysProvider = makeGaugeProvider({
   name: METRIC_OPERATORS_KEYS_TOTAL,
   help: 'Number of node operators keys',
-  labelNames: ['type'] as const,
+  labelNames: ['type', 'stakingModuleId'] as const,
 });
 
 export const PrometheusKeysApiRequestsProvider = makeHistogramProvider({

src/contracts/deposit/deposit.service.ts

@@ -33,7 +33,7 @@ export class DepositService {
   ) {}
 
   @OneAtTime()
-  public async handleNewBlock({ blockNumber }: BlockData): Promise<void> {
+  public async handleNewBlock(blockNumber: number): Promise<void> {
     if (blockNumber % DEPOSIT_EVENTS_CACHE_UPDATE_BLOCK_RATE !== 0) return;
 
     // The event cache is stored with an N block lag to avoid caching data from uncle blocks

src/guardian/block-guard/block-guard.module.ts

@@ -0,0 +1,11 @@
+import { Module } from '@nestjs/common';
+import { DepositModule } from 'contracts/deposit';
+import { SecurityModule } from 'contracts/security';
+import { BlockGuardService } from './block-guard.service';
+
+@Module({
+  imports: [DepositModule, SecurityModule],
+  providers: [BlockGuardService],
+  exports: [BlockGuardService],
+})
+export class BlockGuardModule {}

src/guardian/block-guard/block-guard.service.ts

@@ -0,0 +1,93 @@
+import { Inject, Injectable, LoggerService } from '@nestjs/common';
+import { WINSTON_MODULE_NEST_PROVIDER } from 'nest-winston';
+
+import { DepositService } from 'contracts/deposit';
+import { SecurityService } from 'contracts/security';
+
+import { BlockData } from '../interfaces';
+
+import { InjectMetric } from '@willsoto/nestjs-prometheus';
+import {
+  METRIC_BLOCK_DATA_REQUEST_DURATION,
+  METRIC_BLOCK_DATA_REQUEST_ERRORS,
+} from 'common/prometheus';
+import { Counter, Histogram } from 'prom-client';
+
+@Injectable()
+export class BlockGuardService {
+  protected lastProcessedStateMeta?: { blockHash: string; blockNumber: number };
+
+  constructor(
+    @Inject(WINSTON_MODULE_NEST_PROVIDER)
+    private logger: LoggerService,
+
+    @InjectMetric(METRIC_BLOCK_DATA_REQUEST_DURATION)
+    private blockRequestsHistogram: Histogram<string>,
+
+    @InjectMetric(METRIC_BLOCK_DATA_REQUEST_ERRORS)
+    private blockErrorsCounter: Counter<string>,
+
+    private depositService: DepositService,
+    private securityService: SecurityService,
+  ) {}
+
+  public isNeedToProcessNewState(newMeta: {
+    blockHash: string;
+    blockNumber: number;
+  }) {
+    const lastMeta = this.lastProcessedStateMeta;
+    if (!lastMeta) return true;
+    if (lastMeta.blockNumber > newMeta.blockNumber) {
+      this.logger.error('Keys-api returns old state', newMeta);
+      return false;
+    }
+    return lastMeta.blockHash !== newMeta.blockHash;
+  }
+
+  public setLastProcessedStateMeta(newMeta: {
+    blockHash: string;
+    blockNumber: number;
+  }) {
+    this.lastProcessedStateMeta = newMeta;
+  }
+
+  /**
+   * Collects data from contracts in one place and by block hash,
+   * to reduce the probability of getting data from different blocks
+   * @returns collected data from the current block
+   */
+  public async getCurrentBlockData({
+    blockNumber,
+    blockHash,
+  }: {
+    blockNumber: number;
+    blockHash: string;
+  }): Promise<BlockData> {
+    try {
+      const endTimer = this.blockRequestsHistogram.startTimer();
+
+      const guardianAddress = this.securityService.getGuardianAddress();
+
+      const [depositRoot, depositedEvents, guardianIndex] = await Promise.all([
+        this.depositService.getDepositRoot({ blockHash }),
+        this.depositService.getAllDepositedEvents(blockNumber, blockHash),
+        this.securityService.getGuardianIndex({ blockHash }),
+        this.depositService.handleNewBlock(blockNumber),
+      ]);
+
+      endTimer();
+
+      return {
+        blockNumber,
+        blockHash,
+        depositRoot,
+        depositedEvents,
+        guardianAddress,
+        guardianIndex,
+      };
+    } catch (error) {
+      this.blockErrorsCounter.inc();
+      throw error;
+    }
+  }
+}

src/guardian/block-guard/index.ts

@@ -0,0 +1,2 @@
+export * from './block-guard.module';
+export * from './block-guard.service';

src/guardian/guardian-message/guardian-message.module.ts

@@ -0,0 +1,10 @@
+import { Module } from '@nestjs/common';
+import { MessagesModule } from 'messages';
+import { GuardianMessageService } from './guardian-message.service';
+
+@Module({
+  imports: [MessagesModule],
+  providers: [GuardianMessageService],
+  exports: [GuardianMessageService],
+})
+export class GuardianMessageModule {}

src/guardian/guardian-message/guardian-message.service.ts

@@ -0,0 +1,95 @@
+import { Inject, Injectable, LoggerService } from '@nestjs/common';
+import { WINSTON_MODULE_NEST_PROVIDER } from 'nest-winston';
+import {
+  MessageDeposit,
+  MessageMeta,
+  MessagePause,
+  MessageRequiredFields,
+  MessagesService,
+  MessageType,
+} from 'messages';
+import { BlockData, StakingModuleData } from '../interfaces';
+import { APP_NAME, APP_VERSION } from 'app.constants';
+
+@Injectable()
+export class GuardianMessageService {
+  constructor(
+    @Inject(WINSTON_MODULE_NEST_PROVIDER)
+    private logger: LoggerService,
+    private messagesService: MessagesService,
+  ) {}
+
+  /**
+   * Sends a ping message to the message broker
+   * @param blockData - collected data from the current block
+   */
+  public async pingMessageBroker(
+    { stakingModuleId }: StakingModuleData,
+    blockData: BlockData,
+  ): Promise<void> {
+    const { blockNumber, guardianIndex, guardianAddress } = blockData;
+
+    await this.sendMessageFromGuardian({
+      type: MessageType.PING,
+      blockNumber,
+      guardianIndex,
+      guardianAddress,
+      stakingModuleId,
+    });
+  }
+
+  /**
+   * Sends a deposit message to the message broker
+   * @param message - MessageDeposit object
+   */
+  public sendDepositMessage(message: Omit<MessageDeposit, 'type'>) {
+    return this.sendMessageFromGuardian({
+      ...message,
+      type: MessageType.DEPOSIT,
+    });
+  }
+
+  /**
+   * Sends a pause message to the message broker
+   * @param message - MessagePause object
+   */
+  public sendPauseMessage(message: Omit<MessagePause, 'type'>) {
+    return this.sendMessageFromGuardian({
+      ...message,
+      type: MessageType.PAUSE,
+    });
+  }
+
+  /**
+   * Adds information about the app to the message
+   * @param message - message object
+   * @returns extended message
+   */
+  public addMessageMetaData<T>(message: T): T & MessageMeta {
+    return {
+      ...message,
+      app: { version: APP_VERSION, name: APP_NAME },
+    };
+  }
+
+  /**
+   * Sends a message to the message broker from the guardian
+   * @param messageData - message object
+   */
+  public async sendMessageFromGuardian<T extends MessageRequiredFields>(
+    messageData: T,
+  ): Promise<void> {
+    if (messageData.guardianIndex == -1) {
+      this.logger.warn(
+        'Your address is not in the Guardian List. The message will not be sent',
+      );
+
+      return;
+    }
+
+    const messageWithMeta = this.addMessageMetaData(messageData);
+
+    this.logger.log('Sending a message to broker', messageData);
+    await this.messagesService.sendMessage(messageWithMeta);
+  }
+}

src/guardian/guardian-message/index.ts

@@ -0,0 +1,2 @@
+export * from './guardian-message.module';
+export * from './guardian-message.service';

src/guardian/guardian-metrics/guardian-metrics.module.ts

@@ -0,0 +1,9 @@
+import { Module } from '@nestjs/common';
+import { GuardianMetricsService } from './guardian-metrics.service';
+
+@Module({
+  imports: [],
+  providers: [GuardianMetricsService],
+  exports: [GuardianMetricsService],
+})
+export class GuardianMetricsModule {}

src/guardian/guardian-metrics/guardian-metrics.service.ts

@@ -0,0 +1,125 @@
+import { Injectable } from '@nestjs/common';
+import { VerifiedDepositEvent } from 'contracts/deposit';
+import { BlockData, StakingModuleData } from '../interfaces';
+import { InjectMetric } from '@willsoto/nestjs-prometheus';
+import {
+  METRIC_VALIDATED_DEPOSITS_TOTAL,
+  METRIC_DEPOSITED_KEYS_TOTAL,
+  METRIC_OPERATORS_KEYS_TOTAL,
+  METRIC_INTERSECTIONS_TOTAL,
+} from 'common/prometheus';
+import { Gauge } from 'prom-client';
+
+@Injectable()
+export class GuardianMetricsService {
+  constructor(
+    @InjectMetric(METRIC_VALIDATED_DEPOSITS_TOTAL)
+    private validatedDepositsCounter: Gauge<string>,
+
+    @InjectMetric(METRIC_DEPOSITED_KEYS_TOTAL)
+    private depositedKeysCounter: Gauge<string>,
+
+    @InjectMetric(METRIC_OPERATORS_KEYS_TOTAL)
+    private operatorsKeysCounter: Gauge<string>,
+
+    @InjectMetric(METRIC_INTERSECTIONS_TOTAL)
+    private intersectionsCounter: Gauge<string>,
+  ) {}
+
+  /**
+   * Collects metrics about keys in the deposit contract and keys of node operators
+   * @param blockData - collected data from the current block
+   */
+  public collectMetrics(
+    stakingModuleData: StakingModuleData,
+    blockData: BlockData,
+  ): void {
+    this.collectValidatingMetrics(stakingModuleData, blockData);
+    this.collectDepositMetrics(stakingModuleData, blockData);
+    this.collectOperatorMetrics(stakingModuleData);
+  }
+
+  /**
+   * Collects metrics about validated deposits
+   * @param blockData - collected data from the current block
+   */
+  public collectValidatingMetrics(
+    { stakingModuleId }: StakingModuleData,
+    blockData: BlockData,
+  ): void {
+    const { depositedEvents } = blockData;
+    const { events } = depositedEvents;
+
+    const valid = events.reduce((sum, { valid }) => sum + (valid ? 1 : 0), 0);
+    const invalid = events.reduce((sum, { valid }) => sum + (valid ? 0 : 1), 0);
+
+    this.validatedDepositsCounter.set(
+      { type: 'valid', stakingModuleId },
+      valid,
+    );
+    this.validatedDepositsCounter.set(
+      { type: 'invalid', stakingModuleId },
+      invalid,
+    );
+  }
+
+  /**
+   * Collects metrics about deposited keys
+   * @param blockData - collected data from the current block
+   */
+  public collectDepositMetrics(
+    { stakingModuleId }: StakingModuleData,
+    blockData: BlockData,
+  ): void {
+    const { depositedEvents } = blockData;
+    const { events } = depositedEvents;
+
+    const depositedKeys = events.map(({ pubkey }) => pubkey);
+    const depositedKeysSet = new Set(depositedKeys);
+    const depositedDubsTotal = depositedKeys.length - depositedKeysSet.size;
+
+    this.depositedKeysCounter.set(
+      { type: 'total', stakingModuleId },
+      depositedKeys.length,
+    );
+    this.depositedKeysCounter.set(
+      { type: 'unique', stakingModuleId },
+      depositedKeysSet.size,
+    );
+    this.depositedKeysCounter.set(
+      { type: 'duplicates', stakingModuleId },
+      depositedDubsTotal,
+    );
+  }
+
+  /**
+   * Collects metrics about operators keys
+   * @param blockData - collected data from the current block
+   */
+  public collectOperatorMetrics(stakingModuleData: StakingModuleData): void {
+    const { unusedKeys, stakingModuleId } = stakingModuleData;
+
+    const operatorsKeysTotal = unusedKeys.length;
+    this.operatorsKeysCounter.set(
+      { type: 'total', stakingModuleId },
+      operatorsKeysTotal,
+    );
+  }
+
+  /**
+   * Collects metrics about keys intersections
+   * @param all - all intersections
+   * @param filtered - all intersections
+   */
+  public collectIntersectionsMetrics(
+    stakingModuleId: number,
+    all: VerifiedDepositEvent[],
+    filtered: VerifiedDepositEvent[],
+  ): void {
+    this.intersectionsCounter.set({ type: 'all', stakingModuleId }, all.length);
+    this.intersectionsCounter.set(
+      { type: 'filtered', stakingModuleId },
+      filtered.length,
+    );
+  }
+}

src/guardian/guardian-metrics/index.ts

@@ -0,0 +1,2 @@
+export * from './guardian-metrics.module';
+export * from './guardian-metrics.service';