Process validator performance data (#346)

* Process validator performance data * fix enum * fix cron while loop * initialize variables * fix data insert * add unit testing files * fix insert data number of columns * implement getIntervalsEpochs * rename test * fix type * parse attestationreward * use math round * parse total rewards * add start and end epoch unit testing * add missing returns att rate per clients * refactor cron (#347) * refactor cron * retry mechanism * Execute cron once every epoch within the first 10% of time of the epoch (#349) * Propose changes * add cron to stop handling --------- Co-authored-by: pablomendezroyo <[email protected]> * skip unit test * use protocol https in teku (#350) * Skip epoch if node is syncing and retry if el is offline (#351) * reorg check node health * relocate * Add unit testing to trackValidatorsPerformance cron module (#352) * Add unit testing to getActiveValidators * Implement missing unit testing * remove only * Add missing test to validators data ingest. WIP (#353) * Add missing test to validators data ingest * add unit testing to getClientsUsedPerIntervalsMap * add unit testing for getAttestationSuccessRatePerClients * Reorg track validators performance cron (#355) * Reorg track validators performance cron * remove unused comment --------- Co-authored-by: Marc Font <[email protected]>
dappnode · Sep 25, 2024 · 3a005d8 · 3a005d8
1 parent 7fe473a
commit 3a005d8
Show file tree

Hide file tree

Showing 35 changed files with 1,585 additions and 325 deletions.
diff --git a/packages/brain/src/index.ts b/packages/brain/src/index.ts
@@ -15,9 +15,9 @@ import { params } from "./params.js";
 import {
   CronJob,
   reloadValidators,
-  trackValidatorsPerformance,
+  trackValidatorsPerformanceCron,
   sendProofsOfValidation,
-  getSecondsToNextEpoch
+  startWithinTenFirstPercentageOfEpoch
 } from "./modules/cron/index.js";
 import { PostgresClient } from "./modules/apiClients/index.js";
 import { brainConfig } from "./modules/config/index.js";
@@ -102,29 +102,24 @@ const proofOfValidationCron = new CronJob(shareCronInterval, () =>
   sendProofsOfValidation(signerApi, brainDb, dappnodeSignatureVerifierApi, shareDataWithDappnode)
 );
 proofOfValidationCron.start();
-const trackValidatorsPerformanceCron = new CronJob(slotsPerEpoch * secondsPerSlot * 1000, () =>
-  // once every epoch
-  trackValidatorsPerformance({
-    brainDb,
-    postgresClient,
-    beaconchainApi,
-    minGenesisTime,
-    secondsPerSlot,
-    executionClient,
-    consensusClient
-  })
+
+// executes once every epoch
+export const trackValidatorsPerformanceCronTask = new CronJob(slotsPerEpoch * secondsPerSlot * 1000, () =>
+  trackValidatorsPerformanceCron({ brainDb, postgresClient, beaconchainApi, executionClient, consensusClient })
 );
-const secondsToNextEpoch = getSecondsToNextEpoch({ minGenesisTime, secondsPerSlot });
-// start the cron within the first minute of an epoch
-// If it remains more than 1 minute then wait for the next epoch (+ 10 seconds of margin)
-if (secondsToNextEpoch > 60) setTimeout(() => trackValidatorsPerformanceCron.start(), (secondsToNextEpoch + 10) * 1000);
-else trackValidatorsPerformanceCron.start();
+startWithinTenFirstPercentageOfEpoch({
+  minGenesisTime,
+  secondsPerSlot,
+  slotsPerEpoch,
+  jobFunction: trackValidatorsPerformanceCronTask
+});
 
 // Graceful shutdown
 function handle(signal: string): void {
   logger.info(`${signal} received. Shutting down...`);
   reloadValidatorsCron.stop();
   proofOfValidationCron.stop();
+  trackValidatorsPerformanceCronTask.stop();
   brainDb.close();
   postgresClient.close().catch((err) => logger.error(`Error closing postgres client`, err)); // postgresClient db connection is the only external resource that needs to be closed
   uiServer.close();

diff --git a/packages/brain/src/modules/apiClients/beaconchain/index.ts b/packages/brain/src/modules/apiClients/beaconchain/index.ts
@@ -13,17 +13,15 @@ import {
   BeaconchainSyncingStatusGetResponse,
   BeaconchainSyncCommitteePostResponse,
   BeaconchainBlockRewardsGetResponse,
-  BeaconchainProposerDutiesGetResponse
+  BeaconchainProposerDutiesGetResponse,
+  BlockId
 } from "./types.js";
 import { StandardApi } from "../standard.js";
 import path from "path";
 import { ApiParams } from "../types.js";
 import { Network } from "@stakingbrain/common";
 import { BeaconchainApiError } from "./error.js";
 
-// TODO: BlockId can also be a simple slot in the form of a string. Is this type still necessary?
-type BlockId = "head" | "genesis" | "finalized" | string | `0x${string}`;
-
 export class BeaconchainApi extends StandardApi {
   private SLOTS_PER_EPOCH: number;
 

diff --git a/packages/brain/src/modules/apiClients/beaconchain/types.ts b/packages/brain/src/modules/apiClients/beaconchain/types.ts
@@ -207,3 +207,5 @@ export interface BeaconchainLivenessPostResponse {
     is_live: boolean;
   }[];
 }
+
+export type BlockId = "head" | "genesis" | "finalized" | string | `0x${string}`;
diff --git a/packages/brain/src/modules/apiClients/postgres/index.ts b/packages/brain/src/modules/apiClients/postgres/index.ts
@@ -32,7 +32,7 @@ enum Columns {
   liveness = "liveness",
   blockProposalStatus = "block_proposal_status",
   syncCommitteeRewards = "sync_comittee_rewards",
-  attestationsRewards = "attestations_rewards",
+  attestationsTotalRewards = "attestations_total_rewards",
   error = "error"
 }
 
@@ -123,9 +123,9 @@ CREATE TABLE IF NOT EXISTS ${this.tableName} (
   ${Columns.consensusClient} ${this.CONSENSUS_CLIENT} NOT NULL,
   ${Columns.slot} BIGINT,
   ${Columns.liveness} BOOLEAN,
-  ${Columns.blockProposalStatus} ${this.BLOCK_PROPOSAL_STATUS},
+  ${Columns.blockProposalStatus} ${this.BLOCK_PROPOSAL_STATUS} NOT NULL,
   ${Columns.syncCommitteeRewards} BIGINT,
-  ${Columns.attestationsRewards} JSONB,
+  ${Columns.attestationsTotalRewards} JSONB NOT NULL,
   ${Columns.error} TEXT,
   PRIMARY KEY (${Columns.validatorIndex}, ${Columns.epoch})
 );
@@ -150,12 +150,12 @@ CREATE TABLE IF NOT EXISTS ${this.tableName} (
    * Inserts the given performance data into the database.
    *
    * @param data - The performance data to insert.
-   * @example insertPerformanceData({ validatorIndex: 1, epoch: 1, slot: 1, liveness: true, blockProposalStatus: "missed", syncCommitteeRewards: 100, attestationsRewards: { attestation1: 10, attestation2: 20 } })
+   * @example insertPerformanceData({ validatorIndex: 1, epoch: 1, slot: 1, liveness: true, blockProposalStatus: "missed", syncCommitteeRewards: 100, attestationsTotalRewards: { attestation1: 10, attestation2: 20 } })
    */
   public async insertPerformanceData(data: ValidatorPerformance): Promise<void> {
     const query = `
-INSERT INTO ${this.tableName} (${Columns.validatorIndex}, ${Columns.epoch}, ${Columns.slot}, ${Columns.liveness}, ${Columns.blockProposalStatus}, ${Columns.syncCommitteeRewards}, ${Columns.attestationsRewards}, ${Columns.error})
-VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
+INSERT INTO ${this.tableName} (${Columns.validatorIndex}, ${Columns.epoch}, ${Columns.executionClient}, ${Columns.consensusClient}, ${Columns.slot}, ${Columns.liveness}, ${Columns.blockProposalStatus}, ${Columns.syncCommitteeRewards}, ${Columns.attestationsTotalRewards}, ${Columns.error})
+VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
     `;
 
     await this.sql.unsafe(query, [
@@ -165,15 +165,16 @@ VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
       data.consensusClient,
       data.slot ?? null,
       data.liveness ?? null,
-      data.blockProposalStatus ?? null,
+      data.blockProposalStatus,
       data.syncCommitteeRewards ?? null,
-      JSON.stringify(data.attestationsRewards) ?? null, // JSONB expects a string
+      JSON.stringify(data.attestationsTotalRewards), // JSONB expects a string
       data.error ?? null
     ]);
   }
 
   /**
-   * Get the validators data for the given validator indexes from all epochs.
+   * Get the validators data for the given validator indexes from all epochs. In order to improve data process
+   * it will return a map with the validator index as key and the performance data as value.
    *
    * @param validatorIndexes - The indexes of the validators to get the data for.
    * @returns The performance data for the given validators.
@@ -185,7 +186,7 @@ WHERE ${Columns.validatorIndex} = ANY($1)
     `;
 
     const result = await this.sql.unsafe(query, [validatorIndexes]);
-    // TODO: add type for result
+    // TODO: add type to result
     // eslint-disable-next-line @typescript-eslint/no-explicit-any
     return result.map((row: any) => ({
       validatorIndex: row.validator_index,
@@ -196,11 +197,64 @@ WHERE ${Columns.validatorIndex} = ANY($1)
       liveness: row.liveness,
       blockProposalStatus: row.block_proposal_status,
       syncCommitteeRewards: row.sync_comittee_rewards,
-      attestationsRewards: row.attestations_rewards,
+      attestationsTotalRewards: JSON.parse(row.attestations_total_rewards),
       error: row.error
     }));
   }
 
+  /**
+   * Get tje validators data for the given validator indexes and an epoch start and end range. In order to improve data process
+   * it will return a map with the validator index as key and the performance data as value.
+   *
+   * @param validatorIndexes - The indexes of the validators to get the data for.
+   * @param startEpoch - The start epoch number.
+   * @param endEpoch - The end epoch number.
+   * @returns The performance data for the given validators.
+   */
+  public async getValidatorsDataMapForEpochRange({
+    validatorIndexes,
+    startEpoch,
+    endEpoch
+  }: {
+    validatorIndexes: string[];
+    startEpoch: number;
+    endEpoch: number;
+  }): Promise<Map<string, ValidatorPerformance[]>> {
+    const query = `
+SELECT * FROM ${this.tableName}
+WHERE ${Columns.validatorIndex} = ANY($1)
+AND ${Columns.epoch} >= $2
+AND ${Columns.epoch} <= $3
+    `;
+
+    const result = await this.sql.unsafe(query, [validatorIndexes, startEpoch, endEpoch]);
+    // TODO: add type to result
+    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    return result.reduce((map: Map<string, ValidatorPerformance[]>, row: any) => {
+      const key = row.validator_index;
+      const performanceData = {
+        validatorIndex: row.validator_index,
+        epoch: row.epoch,
+        executionClient: row.execution_client,
+        consensusClient: row.consensus_client,
+        slot: row.slot,
+        liveness: row.liveness,
+        blockProposalStatus: row.block_proposal_status,
+        syncCommitteeRewards: row.sync_comittee_rewards,
+        attestationsTotalRewards: JSON.parse(row.attestations_total_rewards),
+        error: row.error
+      };
+
+      if (map.has(key)) {
+        map.get(key)?.push(performanceData);
+      } else {
+        map.set(key, [performanceData]);
+      }
+
+      return map;
+    }, new Map<string, ValidatorPerformance[]>());
+  }
+
   /**
    * Method to close the database connection.
    */

diff --git a/packages/brain/src/modules/apiClients/postgres/types.ts b/packages/brain/src/modules/apiClients/postgres/types.ts
@@ -7,15 +7,24 @@ export enum BlockProposalStatus {
   Error = "Error"
 }
 
+export interface AttestationsTotalRewards {
+  validator_index: string;
+  head: string;
+  target: string;
+  source: string;
+  inclusion_delay: string;
+  inactivity: string;
+}
+
 export interface ValidatorPerformance {
   validatorIndex: number;
   epoch: number;
   executionClient: ExecutionClient;
   consensusClient: ConsensusClient;
+  blockProposalStatus: BlockProposalStatus;
+  attestationsTotalRewards: AttestationsTotalRewards;
   slot?: number;
   liveness?: boolean;
-  blockProposalStatus?: BlockProposalStatus;
   syncCommitteeRewards?: number;
-  attestationsRewards?: object;
   error?: string;
 }
diff --git a/packages/brain/src/modules/config/networks/gnosis.ts b/packages/brain/src/modules/config/networks/gnosis.ts
@@ -15,7 +15,7 @@ export const gnosisBrainConfig = (
     consensusClient,
     isMevBoostSet,
     executionClientUrl: "http://execution.gnosis.dncore.dappnode:8545",
-    validatorUrl: "http://validator.gnosis.dncore.dappnode:3500",
+    validatorUrl: `${consensusClient === "teku" ? "https" : "http"}://validator.gnosis.dncore.dappnode:3500`,
     beaconchainUrl: "http:/beacon-chain.gnosis.dncore.dappnode:3500",
     blockExplorerUrl: "https://gnosischa.in",
     signerUrl: "http://web3signer.web3signer-gnosis.dappnode:9000",

diff --git a/packages/brain/src/modules/config/networks/holesky.ts b/packages/brain/src/modules/config/networks/holesky.ts
@@ -15,7 +15,7 @@ export const holeskyBrainConfig = (
     consensusClient,
     isMevBoostSet,
     executionClientUrl: "http://execution.holesky.dncore.dappnode:8545",
-    validatorUrl: "http://validator.holesky.dncore.dappnode:3500",
+    validatorUrl: `${consensusClient === "teku" ? "https" : "http"}://validator.holesky.dncore.dappnode:3500`,
     beaconchainUrl: "http:/beacon-chain.holesky.dncore.dappnode:3500",
     blockExplorerUrl: "https://holesky.beaconcha.in",
     signerUrl: "http://web3signer.web3signer-holesky.dappnode:9000",

diff --git a/packages/brain/src/modules/config/networks/lukso.ts b/packages/brain/src/modules/config/networks/lukso.ts
@@ -15,7 +15,7 @@ export const luksoBrainConfig = (
     consensusClient,
     isMevBoostSet,
     executionClientUrl: "http://execution.lukso.dncore.dappnode:8545",
-    validatorUrl: "http://validator.lukso.dncore.dappnode:3500",
+    validatorUrl: `${consensusClient === "teku" ? "https" : "http"}://validator.lukso.dncore.dappnode:3500`,
     beaconchainUrl: "http:/beacon-chain.lukso.dncore.dappnode:3500",
     blockExplorerUrl: "https://explorer.consensus.mainnet.lukso.network/",
     signerUrl: "http://web3signer.web3signer-lukso.dappnode:9000",

diff --git a/packages/brain/src/modules/config/networks/mainnet.ts b/packages/brain/src/modules/config/networks/mainnet.ts
@@ -15,7 +15,7 @@ export const mainnetBrainConfig = (
     consensusClient,
     isMevBoostSet,
     executionClientUrl: "http://execution.mainnet.dncore.dappnode:8545",
-    validatorUrl: "http://validator.mainnet.dncore.dappnode:3500",
+    validatorUrl: `${consensusClient === "teku" ? "https" : "http"}://validator.mainnet.dncore.dappnode:3500`,
     beaconchainUrl: "http:/beacon-chain.mainnet.dncore.dappnode:3500",
     blockExplorerUrl: "https://beaconcha.in",
     signerUrl: "http://web3signer.web3signer.dappnode:9000",

diff --git a/packages/brain/src/modules/config/networks/prater.ts b/packages/brain/src/modules/config/networks/prater.ts
@@ -15,7 +15,7 @@ export const praterBrainConfig = (
     consensusClient,
     isMevBoostSet,
     executionClientUrl: "http://execution.prater.dncore.dappnode:8545",
-    validatorUrl: "http://validator.prater.dncore.dappnode:3500",
+    validatorUrl: `${consensusClient === "teku" ? "https" : "http"}://validator.prater.dncore.dappnode:3500`,
     beaconchainUrl: "http:/beacon-chain.prater.dncore.dappnode:3500",
     blockExplorerUrl: "https://prater.beaconcha.in",
     signerUrl: "http://web3signer.web3signer-prater.dappnode:9000",

diff --git a/packages/brain/src/modules/cron/index.ts b/packages/brain/src/modules/cron/index.ts
@@ -1,4 +1,7 @@
 export { CronJob } from "./cron.js";
 export { reloadValidators } from "./reloadValidators/index.js";
 export { sendProofsOfValidation } from "./sendProofsOfValidation/index.js";
-export { trackValidatorsPerformance, getSecondsToNextEpoch } from "./trackValidatorsPerformance/index.js";
+export {
+  trackValidatorsPerformanceCron,
+  startWithinTenFirstPercentageOfEpoch
+} from "./trackValidatorsPerformance/index.js";
diff --git a/packages/brain/src/modules/cron/trackValidatorsPerformance/checkNodeHealth.ts b/packages/brain/src/modules/cron/trackValidatorsPerformance/checkNodeHealth.ts
diff --git a/...ges/brain/src/modules/cron/trackValidatorsPerformance/getActiveValidatorsLoadedInBrain.ts b/...ges/brain/src/modules/cron/trackValidatorsPerformance/getActiveValidatorsLoadedInBrain.ts
@@ -16,24 +16,22 @@ import { logPrefix } from "./logPrefix.js";
  */
 export async function getActiveValidatorsLoadedInBrain({
   beaconchainApi,
-  brainDb,
-  activeValidatorsIndexes
+  brainDb
 }: {
   beaconchainApi: BeaconchainApi;
   brainDb: BrainDataBase;
-  activeValidatorsIndexes: string[];
-}): Promise<void> {
+}): Promise<string[]> {
   const validatorIndexes = await getValidatorIndexesAndSaveInDb({ beaconchainApi, brainDb });
-  if (validatorIndexes.length === 0) return;
-  (
+  if (validatorIndexes.length === 0) return [];
+  return (
     await beaconchainApi.postStateValidators({
       body: {
         ids: validatorIndexes,
         statuses: validatorIndexes.map(() => ValidatorStatus.ACTIVE_ONGOING)
       },
       stateId: "finalized"
     })
-  ).data.forEach((validator) => activeValidatorsIndexes.push(validator.index));
+  ).data.map((validator) => validator.index.toString());
 }
 
 /**

diff --git a/packages/brain/src/modules/cron/trackValidatorsPerformance/getAttestationsTotalRewards.ts b/packages/brain/src/modules/cron/trackValidatorsPerformance/getAttestationsTotalRewards.ts
@@ -6,24 +6,22 @@ import { TotalRewards } from "../../apiClients/types.js";
  *
  * @param {BeaconchainApi} beaconchainApi - Beaconchain API client.
  * @param {string} epoch - The epoch to get the rewards.
- * @param {string[]} validatorIndexes - Array of validator indexes.
+ * @param {string[]} activeValidatorsIndexes - Array of active validator indexes.
  * @returns {TotalRewards[]} - Array of total rewards for the validators.
  */
 export async function getAttestationsTotalRewards({
   beaconchainApi,
   epoch,
-  validatorIndexes,
-  totalRewards
+  activeValidatorsIndexes
 }: {
   beaconchainApi: BeaconchainApi;
   epoch: string;
-  validatorIndexes: string[];
-  totalRewards: TotalRewards[];
-}): Promise<void> {
-  (
+  activeValidatorsIndexes: string[];
+}): Promise<TotalRewards[]> {
+  return (
     await beaconchainApi.getAttestationsRewards({
       epoch,
-      pubkeysOrIndexes: validatorIndexes
+      pubkeysOrIndexes: activeValidatorsIndexes
     })
-  ).data.total_rewards.forEach((reward) => totalRewards.push(reward));
+  ).data.total_rewards;
 }