feat: daemon

src/common/handlers/handlers.service.ts

@@ -23,15 +23,22 @@ export class HandlersService {
     const payload = await this.buildProvePayload(slashings, withdrawals);
     // TODO: ask before sending if CLI or daemon in watch mode
     await this.sendProves(payload);
+    this.logger.log(`🏁 Proves sent. Root [${blockRoot}]`);
   }
 
   private async buildProvePayload(slashings: string[], withdrawals: string[]): Promise<any> {
     // TODO: implement
     //  this.consensus.getState(...)
+    if (slashings.length || withdrawals.length) {
+      this.logger.warn(`📦 Prove payload: slashings [${slashings}], withdrawals [${withdrawals}]`);
+    }
     return {};
   }
   private async sendProves(payload: any): Promise<void> {
     // TODO: implement
+    if (payload) {
+      this.logger.warn(`📡 Sending proves`);
+    }
   }
 
   private async getUnprovenSlashings(
@@ -114,7 +121,7 @@ export class HandlersService {
   ): string[] {
     const fullWithdrawals = [];
     const blockEpoch = Number(blockInfo.message.slot) / 32;
-    const withdrawals = blockInfo.message.body.execution_payload.withdrawals;
+    const withdrawals = blockInfo.message.body.execution_payload?.withdrawals ?? [];
     for (const withdrawal of withdrawals) {
       const keyInfo = keyInfoFn(Number(withdrawal.validator_index));
       if (keyInfo && blockEpoch >= keyInfo.withdrawableEpoch) {

src/daemon/daemon.module.ts

@@ -1,20 +1,19 @@
 import { Module } from '@nestjs/common';
 
 import { DaemonService } from './daemon.service';
+import { KeysIndexer } from './services/keys-indexer';
+import { RootsProcessor } from './services/roots-processor';
+import { RootsProvider } from './services/roots-provider';
+import { RootsStack } from './services/roots-stack';
 import { ConfigModule } from '../common/config/config.module';
 import { HandlersModule } from '../common/handlers/handlers.module';
 import { LoggerModule } from '../common/logger/logger.module';
 import { PrometheusModule } from '../common/prometheus/prometheus.module';
 import { ProvidersModule } from '../common/providers/providers.module';
 
 @Module({
-  imports: [
-    LoggerModule,
-    ConfigModule,
-    PrometheusModule,
-    ProvidersModule,
-    HandlersModule,
-  ],
-  providers: [DaemonService],
+  imports: [LoggerModule, ConfigModule, PrometheusModule, ProvidersModule, HandlersModule],
+  providers: [DaemonService, KeysIndexer, RootsProvider, RootsProcessor, RootsStack],
+  exports: [DaemonService],
 })
 export class DaemonModule {}

src/daemon/daemon.service.ts

@@ -1,18 +1,50 @@
 import { LOGGER_PROVIDER } from '@lido-nestjs/logger';
-import {
-  Inject,
-  Injectable,
-  LoggerService,
-  OnApplicationBootstrap,
-} from '@nestjs/common';
+import { Inject, Injectable, LoggerService, OnApplicationBootstrap } from '@nestjs/common';
+
+import { KeysIndexer } from './services/keys-indexer';
+import { RootsProcessor } from './services/roots-processor';
+import { RootsProvider } from './services/roots-provider';
+import sleep from './utils/sleep';
+import { ConfigService } from '../common/config/config.service';
+import { Consensus } from '../common/providers/consensus/consensus';
 
 @Injectable()
 export class DaemonService implements OnApplicationBootstrap {
   constructor(
     @Inject(LOGGER_PROVIDER) protected readonly logger: LoggerService,
+    protected readonly config: ConfigService,
+    protected readonly consensus: Consensus,
+    protected readonly keysIndexer: KeysIndexer,
+    protected readonly rootsProvider: RootsProvider,
+    protected readonly rootsProcessor: RootsProcessor,
   ) {}
 
   async onApplicationBootstrap() {
-    this.logger.log('Working mode: DAEMON');
+    this.loop().then();
+  }
+
+  private async loop() {
+    while (true) {
+      try {
+        await this.baseRun();
+      } catch (e) {
+        this.logger.error(e);
+        await sleep(1000);
+      }
+    }
+  }
+
+  private async baseRun() {
+    this.logger.log('🗿 Get finalized header');
+    const header = await this.consensus.getBeaconHeader('finalized');
+    this.logger.log(`💎 Finalized slot [${header.header.message.slot}]. Root [${header.root}]`);
+    await this.keysIndexer.run(header);
+    const nextRoot = await this.rootsProvider.getNext(header);
+    if (nextRoot) {
+      await this.rootsProcessor.process(nextRoot);
+    } else {
+      this.logger.log(`💤 Wait for the next finalized root`);
+      await sleep(12000);
+    }
   }
 }

src/daemon/services/keys-indexer.ts

@@ -0,0 +1,210 @@
+import { Low } from '@huanshiwushuang/lowdb';
+import { JSONFile } from '@huanshiwushuang/lowdb/node';
+import { LOGGER_PROVIDER } from '@lido-nestjs/logger';
+import { Inject, Injectable, LoggerService, OnApplicationBootstrap } from '@nestjs/common';
+
+import { ConfigService } from '../../common/config/config.service';
+import { KeyInfo } from '../../common/handlers/handlers.service';
+import { Consensus } from '../../common/providers/consensus/consensus';
+import {
+  BlockHeaderResponse,
+  RootHex,
+  Slot,
+  StateValidatorResponse,
+} from '../../common/providers/consensus/response.interface';
+import { Keysapi } from '../../common/providers/keysapi/keysapi';
+
+type Info = {
+  moduleAddress: string;
+  moduleId: number;
+  storageStateSlot: number;
+  lastValidatorsCount: number;
+};
+
+type Storage = {
+  [valIndex: number]: KeyInfo;
+};
+
+@Injectable()
+export class KeysIndexer implements OnApplicationBootstrap {
+  private startedAt: number = 0;
+
+  private info: Low<Info>;
+  private storage: Low<Storage>;
+
+  constructor(
+    @Inject(LOGGER_PROVIDER) protected readonly logger: LoggerService,
+    protected readonly config: ConfigService,
+    protected readonly consensus: Consensus,
+    protected readonly keysapi: Keysapi,
+  ) {}
+
+  public async onApplicationBootstrap(): Promise<void> {
+    await this.initOrReadServiceData();
+  }
+
+  public getKey = (valIndex: number): KeyInfo | undefined => {
+    return this.storage.data[valIndex];
+  };
+
+  public async run(finalizedHeader: BlockHeaderResponse): Promise<unknown> {
+    // At one time only one task should be running
+    if (this.startedAt > 0) {
+      this.logger.warn(`🔑 Keys indexer has been running for ${Date.now() - this.startedAt}ms`);
+      return;
+    }
+    const finalizedSlot = Number(finalizedHeader.header.message.slot);
+    if (this.isNotTimeToRun(finalizedSlot)) {
+      this.logger.log('No need to run keys indexer');
+      return;
+    }
+    this.logger.log(`🔑 Keys indexer is running`);
+    const finalizedStateRoot = finalizedHeader.header.message.state_root;
+    if (this.info.data.storageStateSlot == 0) {
+      await this.baseRun(finalizedStateRoot, finalizedSlot);
+      return;
+    }
+    // We shouldn't wait for task to finish
+    // to avoid block processing if indexing fails or stuck
+    this.startedAt = Date.now();
+    this.baseRun(finalizedStateRoot, finalizedSlot)
+      .catch((e) => this.logger.error(e))
+      .finally(() => (this.startedAt = 0));
+  }
+
+  private async baseRun(stateRoot: RootHex, finalizedSlot: Slot): Promise<void> {
+    this.logger.log(`Get validators. State root [${stateRoot}]`);
+    const validators = await this.consensus.getValidators(stateRoot);
+    this.logger.log(`Total validators count: ${validators.length}`);
+    // TODO: do we need to store already full withdrawn keys ?
+    this.info.data.lastValidatorsCount == 0
+      ? await this.initStorage(validators, finalizedSlot)
+      : await this.updateStorage(validators, finalizedSlot);
+    this.logger.log(`CSM validators count: ${Object.keys(this.storage.data).length}`);
+    this.info.data.storageStateSlot = finalizedSlot;
+    this.info.data.lastValidatorsCount = validators.length;
+    await this.info.write();
+  }
+
+  private async initStorage(validators: StateValidatorResponse[], finalizedSlot: Slot): Promise<void> {
+    this.logger.log(`Init keys data`);
+    const csmKeys = await this.keysapi.getModuleKeys(this.info.data.moduleId);
+    this.keysapi.healthCheck(this.consensus.slotToTimestamp(finalizedSlot), csmKeys.meta);
+    const keysMap = new Map<string, { operatorIndex: number; index: number }>();
+    csmKeys.data.keys.forEach((k: any) => keysMap.set(k.key, { ...k }));
+    for (const v of validators) {
+      const keyInfo = keysMap.get(v.validator.pubkey);
+      if (!keyInfo) continue;
+      this.storage.data[Number(v.index)] = {
+        operatorId: keyInfo.operatorIndex,
+        keyIndex: keyInfo.index,
+        pubKey: v.validator.pubkey,
+        // TODO: bigint?
+        withdrawableEpoch: Number(v.validator.withdrawable_epoch),
+      };
+    }
+    await this.storage.write();
+  }
+
+  private async updateStorage(vals: StateValidatorResponse[], finalizedSlot: Slot): Promise<void> {
+    // TODO: should we think about re-using validator indexes?
+    // TODO: should we think about changing WC for existing old vaidators ?
+    if (vals.length - this.info.data.lastValidatorsCount == 0) {
+      this.logger.log(`No new validators in the state`);
+      return;
+    }
+    vals = vals.slice(this.info.data.lastValidatorsCount);
+    const valKeys = vals.map((v: StateValidatorResponse) => v.validator.pubkey);
+    this.logger.log(`New appeared validators count: ${vals.length}`);
+    const csmKeys = await this.keysapi.findModuleKeys(this.info.data.moduleId, valKeys);
+    this.keysapi.healthCheck(this.consensus.slotToTimestamp(finalizedSlot), csmKeys.meta);
+    this.logger.log(`New appeared CSM validators count: ${csmKeys.data.keys.length}`);
+    for (const csmKey of csmKeys.data.keys) {
+      for (const newVal of vals) {
+        if (newVal.validator.pubkey != csmKey.key) continue;
+        this.storage.data[Number(newVal.index)] = {
+          operatorId: csmKey.operatorIndex,
+          keyIndex: csmKey.index,
+          pubKey: csmKey.key,
+          // TODO: bigint?
+          withdrawableEpoch: Number(newVal.validator.withdrawable_epoch),
+        };
+      }
+    }
+    await this.storage.write();
+  }
+
+  public isNotTimeToRun(finalizedSlot: Slot): boolean {
+    const storageTimestamp = this.consensus.slotToTimestamp(this.info.data.storageStateSlot) * 1000;
+    return (
+      this.info.data.storageStateSlot == finalizedSlot ||
+      this.config.get('KEYS_INDEXER_RUNNING_PERIOD') >= Date.now() - storageTimestamp
+    );
+  }
+
+  public eligibleForAnyDuty(slotNumber: Slot): boolean {
+    return this.eligibleForSlashings(slotNumber) || this.eligibleForFullWithdrawals(slotNumber);
+  }
+
+  public eligibleForEveryDuty(slotNumber: Slot): boolean {
+    const eligibleForSlashings = this.eligibleForSlashings(slotNumber);
+    const eligibleForFullWithdrawals = this.eligibleForFullWithdrawals(slotNumber);
+    if (!eligibleForSlashings)
+      this.logger.warn(
+        '🚨 Current keys indexer data might not be ready to detect slashing. ' +
+          'The root will be processed later again',
+      );
+    if (!eligibleForFullWithdrawals)
+      this.logger.warn(
+        '⚠️ Current keys indexer data might not be ready to detect full withdrawal. ' +
+          'The root will be processed later again',
+      );
+    return eligibleForSlashings && eligibleForFullWithdrawals;
+  }
+
+  private eligibleForSlashings(slotNumber: Slot): boolean {
+    // We are ok with oudated indexer for detection slasing
+    // because of a bunch of delays between deposit and validator appearing
+    // TODO: get constants from node
+    const ETH1_FOLLOW_DISTANCE = 2048; // ~8 hours
+    const EPOCHS_PER_ETH1_VOTING_PERIOD = 64; // ~6.8 hours
+    const safeDelay = ETH1_FOLLOW_DISTANCE + EPOCHS_PER_ETH1_VOTING_PERIOD * 32;
+    if (this.info.data.storageStateSlot >= slotNumber) return true;
+    return slotNumber - this.info.data.storageStateSlot <= safeDelay; // ~14.8 hours
+  }
+
+  private eligibleForFullWithdrawals(slotNumber: Slot): boolean {
+    // We are ok with oudated indexer for detection withdrawal
+    // because of MIN_VALIDATOR_WITHDRAWABILITY_DELAY
+    // TODO: get constants from node
+    const MIN_VALIDATOR_WITHDRAWABILITY_DELAY = 256;
+    const safeDelay = MIN_VALIDATOR_WITHDRAWABILITY_DELAY * 32;
+    if (this.info.data.storageStateSlot >= slotNumber) return true;
+    return slotNumber - this.info.data.storageStateSlot <= safeDelay; // ~27 hours
+  }
+
+  private async initOrReadServiceData() {
+    const defaultInfo: Info = {
+      moduleAddress: this.config.get('LIDO_STAKING_MODULE_ADDRESS'),
+      moduleId: 0,
+      storageStateSlot: 0,
+      lastValidatorsCount: 0,
+    };
+    this.info = new Low<Info>(new JSONFile<Info>('.keys-indexer-info.json'), defaultInfo);
+    this.storage = new Low<Storage>(new JSONFile<Storage>('.keys-indexer-storage.json'), {});
+    await this.info.read();
+    await this.storage.read();
+
+    if (this.info.data.moduleId == 0) {
+      const modules = (await this.keysapi.getModules()).data;
+      const module = modules.find(
+        (m: any) => m.stakingModuleAddress.toLowerCase() === this.info.data.moduleAddress.toLowerCase(),
+      );
+      if (!module) {
+        throw new Error(`Module with address ${this.info.data.moduleAddress} not found`);
+      }
+      this.info.data.moduleId = module.id;
+      await this.info.write();
+    }
+  }
+}

src/daemon/services/roots-processor.ts

@@ -0,0 +1,33 @@
+import { LOGGER_PROVIDER } from '@lido-nestjs/logger';
+import { Inject, Injectable, LoggerService } from '@nestjs/common';
+
+import { KeysIndexer } from './keys-indexer';
+import { RootsStack } from './roots-stack';
+import { HandlersService } from '../../common/handlers/handlers.service';
+import { Consensus } from '../../common/providers/consensus/consensus';
+import { RootHex } from '../../common/providers/consensus/response.interface';
+
+@Injectable()
+export class RootsProcessor {
+  constructor(
+    @Inject(LOGGER_PROVIDER) protected readonly logger: LoggerService,
+    protected readonly consensus: Consensus,
+    protected readonly keysIndexer: KeysIndexer,
+    protected readonly rootsStack: RootsStack,
+    protected readonly handlers: HandlersService,
+  ) {}
+
+  public async process(blockRoot: RootHex): Promise<any> {
+    this.logger.log(`🛃 Root in processing [${blockRoot}]`);
+    const blockInfo = await this.consensus.getBlockInfo(blockRoot);
+    const rootSlot = {
+      blockRoot,
+      slotNumber: Number(blockInfo.message.slot),
+    };
+    const indexerIsOK = this.keysIndexer.eligibleForEveryDuty(rootSlot.slotNumber);
+    if (!indexerIsOK) await this.rootsStack.push(rootSlot); // only new will be pushed
+    await this.handlers.prove(blockRoot, blockInfo, this.keysIndexer.getKey);
+    if (indexerIsOK) await this.rootsStack.purge(blockRoot);
+    await this.rootsStack.setLastProcessed(rootSlot);
+  }
+}