feat: add historical state regen

packages/beacon-node/src/chain/chain.ts

@@ -75,6 +75,7 @@ import {BlockAttributes, produceBlockBody} from "./produceBlock/produceBlockBody
 import {computeNewStateRoot} from "./produceBlock/computeNewStateRoot.js";
 import {BlockInput} from "./blocks/types.js";
 import {SeenAttestationDatas} from "./seenCache/seenAttestationData.js";
+import {HistoricalStateRegen} from "./historicalState/index.js";
 
 /**
  * Arbitrary constants, blobs should be consumed immediately in the same slot they are produced.
@@ -105,6 +106,7 @@ export class BeaconChain implements IBeaconChain {
   readonly regen: QueuedStateRegenerator;
   readonly lightClientServer: LightClientServer;
   readonly reprocessController: ReprocessController;
+  readonly historicalStateRegen?: HistoricalStateRegen;
 
   // Ops pool
   readonly attestationPool: AttestationPool;
@@ -162,6 +164,7 @@ export class BeaconChain implements IBeaconChain {
       eth1,
       executionEngine,
       executionBuilder,
+      historicalStateRegen,
     }: {
       config: BeaconConfig;
       db: IBeaconDb;
@@ -174,6 +177,7 @@ export class BeaconChain implements IBeaconChain {
       eth1: IEth1ForBlockProduction;
       executionEngine: IExecutionEngine;
       executionBuilder?: IExecutionBuilder;
+      historicalStateRegen?: HistoricalStateRegen;
     }
   ) {
     this.opts = opts;
@@ -188,6 +192,7 @@ export class BeaconChain implements IBeaconChain {
     this.eth1 = eth1;
     this.executionEngine = executionEngine;
     this.executionBuilder = executionBuilder;
+    this.historicalStateRegen = historicalStateRegen;
     const signal = this.abortController.signal;
     const emitter = new ChainEventEmitter();
     // by default, verify signatures on both main threads and worker threads
@@ -390,11 +395,16 @@ export class BeaconChain implements IBeaconChain {
         return state && {state, executionOptimistic: isOptimisticBlock(block)};
       }
     } else {
-      // request for finalized state
+      // request for finalized state using historical state regen
+      const stateSerialized = await this.historicalStateRegen?.getHistoricalState(slot);
+      if (!stateSerialized) {
+        return null;
+      }
+      const state = this.config
+        .getForkTypes(slot)
+        .BeaconState.deserialize(stateSerialized) as unknown as BeaconStateAllForks;
 
-      // do not attempt regen, just check if state is already in DB
-      const state = await this.db.stateArchive.get(slot);
-      return state && {state, executionOptimistic: false};
+      return {state, executionOptimistic: false};
     }
   }
 

packages/beacon-node/src/chain/historicalState/getHistoricalState.ts

@@ -0,0 +1,80 @@
+import {
+  BeaconStateAllForks,
+  BeaconStateTransitionMetrics,
+  CachedBeaconStateAllForks,
+  DataAvailableStatus,
+  ExecutionPayloadStatus,
+  PubkeyIndexMap,
+  createCachedBeaconState,
+  stateTransition,
+} from "@lodestar/state-transition";
+import {SignedBeaconBlock} from "@lodestar/types/allForks";
+import {BeaconConfig} from "@lodestar/config";
+import {IBeaconDb} from "../../db/index.js";
+
+export function syncPubkeyCache(state: BeaconStateAllForks, pubkey2index: PubkeyIndexMap): void {
+  // Get the validators sub tree once for all the loop
+  const validators = state.validators;
+
+  const newCount = state.validators.length;
+  for (let i = pubkey2index.size; i < newCount; i++) {
+    const pubkey = validators.getReadonly(i).pubkey;
+    pubkey2index.set(pubkey, i);
+  }
+}
+
+export async function getNearestState(
+  slot: number,
+  config: BeaconConfig,
+  db: IBeaconDb,
+  pubkey2index: PubkeyIndexMap
+): Promise<CachedBeaconStateAllForks> {
+  const states = await db.stateArchive.values({limit: 1, lte: slot, reverse: true});
+  if (!states.length) {
+    throw new Error("No near state found in the database");
+  }
+
+  const state = states[0];
+  syncPubkeyCache(state, pubkey2index);
+
+  return createCachedBeaconState(
+    state,
+    {
+      config,
+      pubkey2index,
+      index2pubkey: [],
+    },
+    {
+      skipSyncPubkeys: true,
+    }
+  );
+}
+
+export function getBlocksBetween(from: number, to: number, db: IBeaconDb): AsyncIterable<SignedBeaconBlock> {
+  return db.blockArchive.valuesStream({gt: from, lte: to});
+}
+
+export async function getHistoricalState(
+  slot: number,
+  config: BeaconConfig,
+  db: IBeaconDb,
+  pubkey2index: PubkeyIndexMap,
+  metrics?: BeaconStateTransitionMetrics
+): Promise<Uint8Array> {
+  let state = await getNearestState(slot, config, db, pubkey2index);
+  for await (const block of getBlocksBetween(state.slot, slot, db)) {
+    state = stateTransition(
+      state,
+      block,
+      {
+        verifyProposer: false,
+        verifySignatures: false,
+        verifyStateRoot: false,
+        executionPayloadStatus: ExecutionPayloadStatus.valid,
+        dataAvailableStatus: DataAvailableStatus.available,
+      },
+      metrics
+    );
+  }
+  return state.serialize();
+}

packages/beacon-node/src/chain/historicalState/index.ts

@@ -0,0 +1,63 @@
+import {ModuleThread, Thread, spawn} from "@chainsafe/threads";
+import {chainConfigToJson} from "@lodestar/config";
+import {LoggerNode} from "@lodestar/logger/node";
+import {
+  HistoricalStateRegenInitModules,
+  HistoricalStateRegenModules,
+  HistoricalStateWorkerApi,
+  HistoricalStateWorkerData,
+} from "./types.js";
+
+/**
+ * HistoricalStateRegen limits the damage from recreating historical states
+ * by running regen in a separate worker thread.
+ */
+export class HistoricalStateRegen implements HistoricalStateWorkerApi {
+  private readonly api: ModuleThread<HistoricalStateWorkerApi>;
+  private readonly logger: LoggerNode;
+
+  constructor(modules: HistoricalStateRegenModules) {
+    this.api = modules.api;
+    this.logger = modules.logger;
+  }
+  static async init(modules: HistoricalStateRegenInitModules): Promise<HistoricalStateRegen> {
+    const workerData: HistoricalStateWorkerData = {
+      chainConfigJson: chainConfigToJson(modules.config),
+      genesisValidatorsRoot: modules.config.genesisValidatorsRoot,
+      genesisTime: modules.opts.genesisTime,
+      maxConcurrency: 1,
+      maxLength: 50,
+      dbLocation: modules.opts.dbLocation,
+      metricsEnabled: Boolean(modules.metrics),
+      loggerOpts: modules.logger.toOpts(),
+    };
+
+    const worker = new Worker("./worker.js", {
+      workerData,
+    } as ConstructorParameters<typeof Worker>[1]);
+
+    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    const api = await spawn<HistoricalStateWorkerApi>(worker, {
+      // A Lodestar Node may do very expensive task at start blocking the event loop and causing
+      // the initialization to timeout. The number below is big enough to almost disable the timeout
+      timeout: 5 * 60 * 1000,
+    });
+
+    return new HistoricalStateRegen({...modules, api});
+  }
+
+  async scrapeMetrics(): Promise<string> {
+    return this.api.scrapeMetrics();
+  }
+
+  async close(): Promise<void> {
+    await this.api.close();
+    this.logger.debug("Terminating historical state worker");
+    await Thread.terminate(this.api);
+    this.logger.debug("Terminated historical state worker");
+  }
+
+  async getHistoricalState(slot: number): Promise<Uint8Array> {
+    return this.api.getHistoricalState(slot);
+  }
+}

packages/beacon-node/src/chain/historicalState/types.ts

@@ -0,0 +1,34 @@
+import {ModuleThread} from "@chainsafe/threads";
+import {BeaconConfig} from "@lodestar/config";
+import {LoggerNode, LoggerNodeOpts} from "@lodestar/logger/node";
+import {Metrics} from "../../metrics/index.js";
+
+export type HistoricalStateRegenInitModules = {
+  opts: {
+    genesisTime: number;
+    dbLocation: string;
+  };
+  config: BeaconConfig;
+  logger: LoggerNode;
+  metrics: Metrics | null;
+};
+export type HistoricalStateRegenModules = HistoricalStateRegenInitModules & {
+  api: ModuleThread<HistoricalStateWorkerApi>;
+};
+
+export type HistoricalStateWorkerData = {
+  chainConfigJson: Record<string, string>;
+  genesisValidatorsRoot: Uint8Array;
+  genesisTime: number;
+  maxConcurrency: number;
+  maxLength: number;
+  dbLocation: string;
+  metricsEnabled: boolean;
+  loggerOpts: LoggerNodeOpts;
+};
+
+export type HistoricalStateWorkerApi = {
+  close(): Promise<void>;
+  scrapeMetrics(): Promise<string>;
+  getHistoricalState(slot: number): Promise<Uint8Array>;
+};