feat: add support for backfilling blocks

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

@@ -233,6 +233,7 @@ export class ArchiveStore {
       const prunedBlocks = this.chain.forkChoice.prune(finalized.rootHex);
       timer?.({source: ArchiveStoreTask.ForkchoicePrune});
 
+      // Todo: Uncomment after integrating backfill db repositories
       timer = this.metrics?.processFinalizedCheckpoint.durationByTask.startTimer();
       await updateBackfillRange({chain: this.chain, db: this.db, logger: this.logger}, finalized);
       timer?.({source: ArchiveStoreTask.UpdateBackfillRange});

packages/beacon-node/src/chain/archiveStore/utils/updateBackfillRange.ts

@@ -1,8 +1,15 @@
+import {Key} from "interface-datastore";
+import {KeyValue} from "@lodestar/db";
 import {CheckpointWithHex} from "@lodestar/fork-choice";
 import {Logger} from "@lodestar/logger";
+import {SLOTS_PER_EPOCH} from "@lodestar/params";
+import {computeEpochAtSlot} from "@lodestar/state-transition";
+import {prettyPrintIndices} from "@lodestar/utils";
 import {IBeaconDb} from "../../../db/interface.js";
+import {BackfillState, EpochBackfillState} from "../../../db/repositories/backfillState.ts";
 import {IBeaconChain} from "../../interface.js";
 
+// Todo: Update comments wrt BackfillRange
 /**
  * Backfill sync relies on verified connected ranges (which are represented as key,value
  * with a verified jump from a key back to value). Since the node could have progressed
@@ -18,31 +25,71 @@ export async function updateBackfillRange(
   finalized: CheckpointWithHex
 ): Promise<void> {
   try {
+    // const {ancestors: finalizedCanonicalBlocks, nonAncestors: finalizedNonCanonicalBlocks} =
+    //   chain.forkChoice.getAllAncestorAndNonAncestorBlocks(finalized.rootHex);
+
     // Mark the sequence in backfill db from finalized block's slot till anchor slot as
     // filled.
     const finalizedBlockFC = chain.forkChoice.getBlockHex(finalized.rootHex);
-    if (finalizedBlockFC && finalizedBlockFC.slot > chain.anchorStateLatestBlockSlot) {
-      await db.backfilledRanges.put(finalizedBlockFC.slot, chain.anchorStateLatestBlockSlot);
-
-      // Clear previously marked sequence till anchorStateLatestBlockSlot, without
-      // touching backfill sync process sequence which are at
-      // <=anchorStateLatestBlockSlot i.e. clear >anchorStateLatestBlockSlot
-      // and < currentSlot
-      const filteredSeqs = await db.backfilledRanges.entries({
-        gt: chain.anchorStateLatestBlockSlot,
-        lt: finalizedBlockFC.slot,
+    const previousBackfillRange = await db.backfillRange.get();
+
+    const finalizedPostDeneb = finalized.epoch >= chain.config.DENEB_FORK_EPOCH;
+    const finalizedPostFulu = finalized.epoch >= chain.config.FULU_FORK_EPOCH;
+
+    if (
+      finalizedBlockFC &&
+      (finalizedBlockFC.slot > chain.anchorStateLatestBlockSlot ||
+        (previousBackfillRange && finalizedBlockFC.slot > previousBackfillRange?.endingEpoch * SLOTS_PER_EPOCH))
+    ) {
+      await db.backfillRange.put({
+        beginningEpoch: computeEpochAtSlot(finalizedBlockFC.slot),
+        endingEpoch: previousBackfillRange?.endingEpoch || computeEpochAtSlot(chain.anchorStateLatestBlockSlot),
       });
-      logger.debug("updated backfilledRanges", {
-        key: finalizedBlockFC.slot,
-        value: chain.anchorStateLatestBlockSlot,
+      // DEBUG_CODE
+      logger.info("Updated backfillRange while migrating from hot to cold db", {
+        beginningEpoch: computeEpochAtSlot(finalizedBlockFC.slot),
+        endingEpoch: previousBackfillRange?.endingEpoch || computeEpochAtSlot(chain.anchorStateLatestBlockSlot),
+        previousBackfillRangeBeginningEpoch: previousBackfillRange?.beginningEpoch,
+        previousBackfillRangeEndingEpoch: previousBackfillRange?.endingEpoch,
+        chainAnchorStateLatestBlockSlotEpoch: computeEpochAtSlot(chain.anchorStateLatestBlockSlot),
+      });
+      // DEBUG_CODE
+
+      // const custodyColumns = chain.custodyConfig.custodyColumns;
+      await db.backfillState.put(finalized.epoch, {
+        hasBlock: true,
+        // check if blobs & columns are filled in live chain
+        hasBlobs: finalizedPostDeneb ? true : null,
+        columnIndices: finalizedPostFulu ? [] : null,
       });
-      if (filteredSeqs.length > 0) {
-        await db.backfilledRanges.batchDelete(filteredSeqs.map((entry) => entry.key));
-        logger.debug(
-          `Forward Sync - cleaned up backfilledRanges between ${finalizedBlockFC.slot},${chain.anchorStateLatestBlockSlot}`,
-          {seqs: JSON.stringify(filteredSeqs)}
-        );
-      }
+
+      // DEBUG_CODE
+      logger.info("Updated backfillState while migrating from hot to cold db", {
+        finalizedEpoch: finalized.epoch,
+        hasBlock: true,
+        hasBlobs: finalizedPostDeneb ? true : null,
+        columnIndices: finalizedPostFulu ? prettyPrintIndices([]) : null,
+      });
+      // DEBUG_CODE
+
+      // Todo: verify if this function runs every epoch, else intermediate epoch backfill states will be empty.
+      // Below could be a possible solution to this issue.
+
+      // // In case of long unfinality, this needs to be done to save multiple epochs
+      // // First, find all *unique* epochs from the list of finalized blocks
+      // const uniqueEpochs = Array.from(new Set(finalizedCanonicalBlocks.map((block) => block.finalizedEpoch)));
+      // const backfillStates: KeyValue<number, EpochBackfillState>[] = uniqueEpochs.map((epoch) => {
+      //   return {
+      //     key: epoch,
+      //     value: {
+      //       hasBlock: true,
+      //       // check if blobs & columns are filled in live chain
+      //       hasBlobs: finalizedPostDeneb ? true : null,
+      //       columnIndices: finalizedPostFulu ? [] : null,
+      //     },
+      //   };
+      // });
+      // await db.backfillState.batchPut(backfillStates);
     }
   } catch (e) {
     logger.error("Error updating backfilledRanges on finalization", {epoch: finalized.epoch}, e as Error);

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

@@ -5,7 +5,7 @@ import {CheckpointStateRepository} from "./repositories/checkpointState.js";
 import {
   AttesterSlashingRepository,
   BLSToExecutionChangeRepository,
-  BackfilledRanges,
+  BackfillState,
   BestLightClientUpdateRepository,
   BlobSidecarsArchiveRepository,
   BlobSidecarsRepository,
@@ -23,7 +23,7 @@ import {
   SyncCommitteeWitnessRepository,
   VoluntaryExitRepository,
 } from "./repositories/index.js";
-import {PreGenesisState, PreGenesisStateLastProcessedBlock} from "./single/index.js";
+import {BackfillRange, PreGenesisState, PreGenesisStateLastProcessedBlock} from "./single/index.js";
 
 export type BeaconDbModules = {
   config: ChainForkConfig;
@@ -59,7 +59,8 @@ export class BeaconDb implements IBeaconDb {
   syncCommittee: SyncCommitteeRepository;
   syncCommitteeWitness: SyncCommitteeWitnessRepository;
 
-  backfilledRanges: BackfilledRanges;
+  backfillRange: BackfillRange;
+  backfillState: BackfillState;
 
   constructor(
     config: ChainForkConfig,
@@ -92,7 +93,8 @@ export class BeaconDb implements IBeaconDb {
     this.syncCommittee = new SyncCommitteeRepository(config, db);
     this.syncCommitteeWitness = new SyncCommitteeWitnessRepository(config, db);
 
-    this.backfilledRanges = new BackfilledRanges(config, db);
+    this.backfillRange = new BackfillRange(config, db);
+    this.backfillState = new BackfillState(config, db);
   }
 
   close(): Promise<void> {

packages/beacon-node/src/db/interface.ts

@@ -3,7 +3,7 @@ import {CheckpointStateRepository} from "./repositories/checkpointState.js";
 import {
   AttesterSlashingRepository,
   BLSToExecutionChangeRepository,
-  BackfilledRanges,
+  BackfillState,
   BestLightClientUpdateRepository,
   BlobSidecarsArchiveRepository,
   BlobSidecarsRepository,
@@ -21,7 +21,7 @@ import {
   SyncCommitteeWitnessRepository,
   VoluntaryExitRepository,
 } from "./repositories/index.js";
-import {PreGenesisState, PreGenesisStateLastProcessedBlock} from "./single/index.js";
+import {BackfillRange, PreGenesisState, PreGenesisStateLastProcessedBlock} from "./single/index.js";
 
 /**
  * The DB service manages the data layer of the beacon chain
@@ -65,7 +65,8 @@ export interface IBeaconDb {
   syncCommittee: SyncCommitteeRepository;
   syncCommitteeWitness: SyncCommitteeWitnessRepository;
 
-  backfilledRanges: BackfilledRanges;
+  backfillState: BackfillState;
+  backfillRange: BackfillRange;
 
   pruneHotDb(): Promise<void>;
 

packages/beacon-node/src/db/repositories/index.ts

@@ -1,5 +1,4 @@
 export {AttesterSlashingRepository} from "./attesterSlashing.js";
-export {BackfilledRanges} from "./backfilledRanges.js";
 export {BackfillState} from "./backfillState.js";
 export {BlobSidecarsRepository} from "./blobSidecars.js";
 export {BlobSidecarsArchiveRepository} from "./blobSidecarsArchive.js";

packages/beacon-node/src/db/single/backfillRange.ts

@@ -1,6 +1,6 @@
 import {ContainerType, Type, ValueOf} from "@chainsafe/ssz";
 import {ChainForkConfig} from "@lodestar/config";
-import {Db, DbReqOpts, encodeKey as _encodeKey} from "@lodestar/db";
+import {Db, DbReqOpts} from "@lodestar/db";
 import {ssz} from "@lodestar/types";
 import {Bucket, getBucketNameByValue} from "../buckets.js";
 
@@ -17,6 +17,20 @@ export type BackfillRangeWrapper = ValueOf<typeof backfillRangeSSZ>;
 // bucket which stores Epoch -> EpochBackfillState key value pairs
 export const BACKFILL_RANGE_KEY = -1;
 
+// Note: intToBytes(-1) incorrectly encodes to 0x0000000000000001, so we manually
+// encode as 0xFFFFFFFFFFFFFFFF (two's complement big-endian representation of -1)
+// This ensures no collision with epoch keys
+const BACKFILL_RANGE_KEY_BYTES = new Uint8Array([0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff]);
+const BUCKET_PREFIX_LEN = 1;
+const KEY_LEN = 8;
+
+function encodeBackfillRangeKey(bucket: number): Uint8Array {
+  const buf = new Uint8Array(BUCKET_PREFIX_LEN + KEY_LEN);
+  buf[0] = bucket;
+  buf.set(BACKFILL_RANGE_KEY_BYTES, BUCKET_PREFIX_LEN);
+  return buf;
+}
+
 export class BackfillRange {
   private readonly bucket: Bucket;
   private readonly db: Db;
@@ -27,7 +41,7 @@ export class BackfillRange {
   constructor(_config: ChainForkConfig, db: Db) {
     this.db = db;
     this.bucket = Bucket.backfill_state;
-    this.key = _encodeKey(this.bucket, BACKFILL_RANGE_KEY);
+    this.key = encodeBackfillRangeKey(this.bucket);
     this.type = backfillRangeSSZ;
     this.dbReqOpts = {bucketId: getBucketNameByValue(this.bucket)};
   }

packages/beacon-node/src/metrics/metrics/lodestar.ts

@@ -18,7 +18,7 @@ import {OpSource} from "../../chain/validatorMonitor.js";
 import {ExecutionPayloadStatus} from "../../execution/index.js";
 import {GossipType} from "../../network/index.js";
 import {CannotAcceptWorkReason, ReprocessRejectReason} from "../../network/processor/index.js";
-import {BackfillSyncMethod} from "../../sync/backfill/backfill.js";
+import {BackfillSyncMethod} from "../../sync/backfill/index.js";
 import {PendingBlockType} from "../../sync/types.js";
 import {PeerSyncType, RangeSyncType} from "../../sync/utils/remoteSyncType.js";
 import {AllocSource} from "../../util/bufferPool.js";