backfill sync from an anchor checkpoint state

packages/cli/src/cmds/beacon/handler.ts

@@ -72,7 +72,14 @@ export async function beaconHandler(args: IBeaconArgs & IGlobalArgs): Promise<vo
 
   // BeaconNode setup
   try {
-    const anchorState = await initBeaconState(options, args, config, db, logger, abortController.signal);
+    const {anchorState, wsCheckpoint} = await initBeaconState(
+      options,
+      args,
+      config,
+      db,
+      logger,
+      abortController.signal
+    );
     const beaconConfig = createIBeaconConfig(config, anchorState.genesisValidatorsRoot);
     const node = await BeaconNode.init({
       opts: options,
@@ -81,6 +88,7 @@ export async function beaconHandler(args: IBeaconArgs & IGlobalArgs): Promise<vo
       logger,
       libp2p: await createNodeJsLibp2p(peerId, options.network, {peerStoreDir: beaconPaths.peerStoreDir}),
       anchorState,
+      wsCheckpoint,
       metricsRegistries,
     });
 

packages/cli/src/cmds/beacon/initBeaconState.ts

@@ -38,7 +38,7 @@ async function initAndVerifyWeakSubjectivityState(
   store: TreeBacked<allForks.BeaconState>,
   wsState: TreeBacked<allForks.BeaconState>,
   wsCheckpoint: Checkpoint
-): Promise<TreeBacked<allForks.BeaconState>> {
+): Promise<{anchorState: TreeBacked<allForks.BeaconState>; wsCheckpoint: Checkpoint}> {
   // Check if the store's state and wsState are compatible
   if (
     store.genesisTime !== wsState.genesisTime ||
@@ -64,7 +64,10 @@ async function initAndVerifyWeakSubjectivityState(
     throw new Error("Fetched weak subjectivity checkpoint not within weak subjectivity period.");
   }
 
-  return await initStateFromAnchorState(config, db, logger, anchorState);
+  anchorState = await initStateFromAnchorState(config, db, logger, anchorState);
+
+  // Return the latest anchorState but still return original wsCheckpoint to validate in backfill
+  return {anchorState, wsCheckpoint};
 }
 
 /**
@@ -83,7 +86,7 @@ export async function initBeaconState(
   db: IBeaconDb,
   logger: ILogger,
   signal: AbortSignal
-): Promise<TreeBacked<allForks.BeaconState>> {
+): Promise<{anchorState: TreeBacked<allForks.BeaconState>; wsCheckpoint?: Checkpoint}> {
   // fetch the latest state stored in the db
   // this will be used in all cases, if it exists, either used during verification of a weak subjectivity state, or used directly as the anchor state
   const lastDbState = await db.stateArchive.lastValue();
@@ -132,16 +135,19 @@ export async function initBeaconState(
   } else if (lastDbState) {
     // start the chain from the latest stored state in the db
     const config = createIBeaconConfig(chainForkConfig, lastDbState.genesisValidatorsRoot);
-    return await initStateFromAnchorState(config, db, logger, lastDbState);
+    const anchorState = await initStateFromAnchorState(config, db, logger, lastDbState);
+    return {anchorState};
   } else {
     const genesisStateFile = args.genesisStateFile || getGenesisFileUrl(args.network || defaultNetwork);
     if (genesisStateFile && !args.forceGenesis) {
       const stateBytes = await downloadOrLoadFile(genesisStateFile);
-      const anchorState = getStateTypeFromBytes(chainForkConfig, stateBytes).createTreeBackedFromBytes(stateBytes);
+      let anchorState = getStateTypeFromBytes(chainForkConfig, stateBytes).createTreeBackedFromBytes(stateBytes);
       const config = createIBeaconConfig(chainForkConfig, anchorState.genesisValidatorsRoot);
-      return await initStateFromAnchorState(config, db, logger, anchorState);
+      anchorState = await initStateFromAnchorState(config, db, logger, anchorState);
+      return {anchorState};
     } else {
-      return await initStateFromEth1({config: chainForkConfig, db, logger, opts: options.eth1, signal});
+      const anchorState = await initStateFromEth1({config: chainForkConfig, db, logger, opts: options.eth1, signal});
+      return {anchorState};
     }
   }
 }

packages/db/src/schema.ts

@@ -66,6 +66,8 @@ export enum Bucket {
   lightClient_bestPartialLightClientUpdate = 54, // SyncPeriod -> PartialLightClientUpdate
 
   validator_metaData = 41,
+
+  backfilled_ranges = 42, // Backfilled From to To, inclusive of both From, To
 }
 
 export enum Key {

packages/lodestar/src/chain/archiver/archiveStates.ts

@@ -3,6 +3,7 @@
  */
 
 import {ILogger} from "@chainsafe/lodestar-utils";
+import {Slot} from "@chainsafe/lodestar-types";
 import {computeEpochAtSlot} from "@chainsafe/lodestar-beacon-state-transition";
 import {IBeaconDb} from "../../db";
 import {CheckpointStateCache} from "../stateCache";
@@ -42,10 +43,22 @@ export class StatesArchiver {
    * epoch - 1024*2    epoch - 1024    epoch - 32    epoch
    * ```
    */
-  async maybeArchiveState(finalized: CheckpointWithHex): Promise<void> {
+  async maybeArchiveState(finalized: CheckpointWithHex, anchorSlot: Slot): Promise<void> {
     const lastStoredSlot = await this.db.stateArchive.lastKey();
     const lastStoredEpoch = computeEpochAtSlot(lastStoredSlot ?? 0);
 
+    // Mark the sequence in backfill db from finalized slot till anchor slot as filled
+    const finalizedState = this.checkpointStateCache.get(finalized);
+    if (!finalizedState) {
+      throw Error("No state in cache for finalized checkpoint state epoch #" + finalized.epoch);
+    }
+    await this.db.backfilledRanges.put(finalizedState.slot, anchorSlot);
+
+    // Clear previously marked sequence till anchorSlot, without touching backfill sync
+    // process sequence which are at <=anchorSlot i.e. clear >anchorSlot and < currentSlot
+    const filteredSeqs = await this.db.backfilledRanges.keys({gt: anchorSlot, lt: finalizedState.slot});
+    await this.db.backfilledRanges.batchDelete(filteredSeqs);
+
     if (finalized.epoch - lastStoredEpoch > PERSIST_TEMP_STATE_EVERY_EPOCHS) {
       await this.archiveState(finalized);
 

packages/lodestar/src/chain/archiver/index.ts

@@ -71,7 +71,7 @@ export class Archiver {
       await archiveBlocks(this.db, this.chain.forkChoice, this.chain.lightClientServer, this.logger, finalized);
 
       // should be after ArchiveBlocksTask to handle restart cleanly
-      await this.statesArchiver.maybeArchiveState(finalized);
+      await this.statesArchiver.maybeArchiveState(finalized, this.chain.anchorSlot);
 
       await Promise.all([
         this.chain.checkpointStateCache.pruneFinalized(finalizedEpoch),

packages/lodestar/src/chain/chain.ts

@@ -51,6 +51,7 @@ export class BeaconChain implements IBeaconChain {
   readonly executionEngine: IExecutionEngine;
   // Expose config for convenience in modularized functions
   readonly config: IBeaconConfig;
+  readonly anchorSlot: Slot;
 
   bls: IBlsVerifier;
   forkChoice: IForkChoice;
@@ -109,6 +110,7 @@ export class BeaconChain implements IBeaconChain {
     this.logger = logger;
     this.metrics = metrics;
     this.genesisTime = anchorState.genesisTime;
+    this.anchorSlot = anchorState.slot;
     this.genesisValidatorsRoot = anchorState.genesisValidatorsRoot.valueOf() as Uint8Array;
     this.eth1 = eth1;
     this.executionEngine = executionEngine;

packages/lodestar/src/chain/interface.ts

@@ -40,6 +40,9 @@ export interface IBeaconChain {
   // Expose config for convenience in modularized functions
   readonly config: IBeaconConfig;
 
+  /** The initial slot that the chain is started with */
+  readonly anchorSlot: Slot;
+
   bls: IBlsVerifier;
   forkChoice: IForkChoice;
   clock: IBeaconClock;

packages/lodestar/src/constants/network.ts

@@ -44,7 +44,6 @@ export type RpcResponseStatusError = Exclude<RespStatus, RespStatus.SUCCESS>;
 export const GOSSIP_MAX_SIZE = 2 ** 20;
 /** The maximum allowed size of uncompressed req/resp chunked responses. */
 export const MAX_CHUNK_SIZE = 2 ** 20;
-
 /** The maximum time to wait for first byte of request response (time-to-first-byte). */
 export const TTFB_TIMEOUT = 5 * 1000; // 5 sec
 /** The maximum time for complete response transfer. */

packages/lodestar/src/db/beacon.ts

@@ -18,6 +18,7 @@ import {
   CheckpointHeaderRepository,
   SyncCommitteeRepository,
   SyncCommitteeWitnessRepository,
+  BackfilledRanges,
 } from "./repositories";
 import {PreGenesisState, PreGenesisStateLastProcessedBlock} from "./single";
 
@@ -44,6 +45,8 @@ export class BeaconDb extends DatabaseService implements IBeaconDb {
   syncCommittee: SyncCommitteeRepository;
   syncCommitteeWitness: SyncCommitteeWitnessRepository;
 
+  backfilledRanges: BackfilledRanges;
+
   constructor(opts: IDatabaseApiOptions) {
     super(opts);
     this.metrics = opts.metrics;
@@ -65,6 +68,8 @@ export class BeaconDb extends DatabaseService implements IBeaconDb {
     this.checkpointHeader = new CheckpointHeaderRepository(this.config, this.db, this.metrics);
     this.syncCommittee = new SyncCommitteeRepository(this.config, this.db, this.metrics);
     this.syncCommitteeWitness = new SyncCommitteeWitnessRepository(this.config, this.db, this.metrics);
+
+    this.backfilledRanges = new BackfilledRanges(this.config, this.db, this.metrics);
   }
 
   async stop(): Promise<void> {

packages/lodestar/src/db/interface.ts

@@ -18,6 +18,7 @@ import {
   CheckpointHeaderRepository,
   SyncCommitteeRepository,
   SyncCommitteeWitnessRepository,
+  BackfilledRanges,
 } from "./repositories";
 import {PreGenesisState, PreGenesisStateLastProcessedBlock} from "./single";
 
@@ -58,6 +59,8 @@ export interface IBeaconDb {
   syncCommittee: SyncCommitteeRepository;
   syncCommitteeWitness: SyncCommitteeWitnessRepository;
 
+  backfilledRanges: BackfilledRanges;
+
   /**
    * Start the connection to the db instance and open the db store.
    */

packages/lodestar/src/db/repositories/backfilledRanges.ts

@@ -0,0 +1,28 @@
+import {IChainForkConfig} from "@chainsafe/lodestar-config";
+import {Slot, ssz} from "@chainsafe/lodestar-types";
+import {IDatabaseController, Bucket, IDbMetrics, Repository} from "@chainsafe/lodestar-db";
+import {bytesToInt} from "@chainsafe/lodestar-utils";
+
+/**
+ * Slot to slot ranges that ensure that block range is fully backfilled
+ *
+ * If node starts backfilling at slots 1000, and backfills to 800, there will be an entry
+ * 1000 -> 800
+ *
+ * When the node is backfilling if it starts at 1200 and backfills to 1000, it will find this sequence and,
+ * jump directly to 800 and delete the key 1000.
+ */
+export class BackfilledRanges extends Repository<Slot, Slot> {
+  constructor(config: IChainForkConfig, db: IDatabaseController<Uint8Array, Uint8Array>, metrics?: IDbMetrics) {
+    super(config, db, Bucket.backfilled_ranges, ssz.Slot, metrics);
+  }
+
+  decodeKey(data: Buffer): number {
+    return bytesToInt((super.decodeKey(data) as unknown) as Uint8Array, "be");
+  }
+
+  // eslint-disable-next-line @typescript-eslint/no-unused-vars
+  getId(value: Slot): number {
+    throw new Error("Cannot get the db key from slot");
+  }
+}

packages/lodestar/src/db/repositories/index.ts

@@ -14,3 +14,4 @@ export {BestPartialLightClientUpdateRepository} from "./lightclientBestPartialUp
 export {CheckpointHeaderRepository} from "./lightclientCheckpointHeader";
 export {SyncCommitteeRepository} from "./lightclientSyncCommittee";
 export {SyncCommitteeWitnessRepository} from "./lightclientSyncCommitteeWitness";
+export {BackfilledRanges} from "./backfilledRanges";

packages/lodestar/src/metrics/metrics/lodestar.ts

@@ -426,6 +426,30 @@ export function createLodestarMetrics(
       }),
     },
 
+    backfillSync: {
+      backfilledTillSlot: register.gauge({
+        name: "lodestar_backfill_till_slot",
+        help: "Current lowest backfilled slot",
+      }),
+      prevFinOrWsSlot: register.gauge({
+        name: "lodestar_backfill_prev_fin_or_ws_slot",
+        help: "Slot of previous finalized or wsCheckpoint block to be validated",
+      }),
+      totalBlocks: register.gauge<"method">({
+        name: "lodestar_backfill_sync_blocks_total",
+        help: "Total amount of backfilled blocks",
+        labelNames: ["method"],
+      }),
+      errors: register.gauge({
+        name: "lodestar_backfill_sync_errors_total",
+        help: "Total number of errors while backfilling",
+      }),
+      status: register.gauge({
+        name: "lodestar_backfill_sync_status",
+        help: "Current backfill syncing status: [Aborted, Pending, Syncing, Completed]",
+      }),
+    },
+
     // Validator monitoring
 
     validatorMonitor: {

packages/lodestar/src/network/interface.ts

@@ -8,7 +8,7 @@ import PeerId from "peer-id";
 import {INetworkEventBus} from "./events";
 import {Eth2Gossipsub} from "./gossip";
 import {MetadataController} from "./metadata";
-import {IPeerRpcScoreStore, IPeerMetadataStore} from "./peers";
+import {IPeerRpcScoreStore, IPeerMetadataStore, PeerAction} from "./peers";
 import {IReqResp} from "./reqresp";
 import {IAttnetsService, ISubnetsService, CommitteeSubscription} from "./subnets";
 
@@ -39,6 +39,7 @@ export interface INetwork {
   /** Subscribe, search peers, join long-lived syncnets */
   prepareSyncCommitteeSubnets(subscriptions: CommitteeSubscription[]): void;
   reStatusPeers(peers: PeerId[]): void;
+  reportPeer(peer: PeerId, action: PeerAction, actionName?: string): void;
 
   // Gossip handler
   subscribeGossipCoreTopics(): void;

packages/lodestar/src/network/network.ts

@@ -22,7 +22,7 @@ import {MetadataController} from "./metadata";
 import {getActiveForks, getCurrentAndNextFork, FORK_EPOCH_LOOKAHEAD} from "./forks";
 import {IPeerMetadataStore, Libp2pPeerMetadataStore} from "./peers/metastore";
 import {PeerManager} from "./peers/peerManager";
-import {IPeerRpcScoreStore, PeerRpcScoreStore} from "./peers";
+import {IPeerRpcScoreStore, PeerAction, PeerRpcScoreStore} from "./peers";
 import {INetworkEventBus, NetworkEventBus} from "./events";
 import {AttnetsService, SyncnetsService, CommitteeSubscription} from "./subnets";
 
@@ -205,6 +205,10 @@ export class Network implements INetwork {
     this.peerManager.reStatusPeers(peers);
   }
 
+  reportPeer(peer: PeerId, action: PeerAction, actionName?: string): void {
+    this.peerRpcScores.applyAction(peer, action, actionName);
+  }
+
   /**
    * Subscribe to all gossip events. Safe to call multiple times
    */