diff --git a/packages/beacon-node/src/chain/archiveStore/utils/updateBackfillRange.ts b/packages/beacon-node/src/chain/archiveStore/utils/updateBackfillRange.ts index 1ae4d5540ace..33bf2e408f7f 100644 --- a/packages/beacon-node/src/chain/archiveStore/utils/updateBackfillRange.ts +++ b/packages/beacon-node/src/chain/archiveStore/utils/updateBackfillRange.ts @@ -17,34 +17,28 @@ export async function updateBackfillRange( {chain, db, logger}: {chain: IBeaconChain; db: IBeaconDb; logger: Logger}, finalized: CheckpointWithHex ): Promise { - try { - // Mark the sequence in backfill db from finalized block's slot till anchor slot as - // filled. - const finalizedBlockFC = chain.forkChoice.getBlockHexDefaultStatus(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, - }); - logger.debug("updated backfilledRanges", { - key: finalizedBlockFC.slot, - value: chain.anchorStateLatestBlockSlot, - }); - 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)} - ); - } - } - } catch (e) { - logger.error("Error updating backfilledRanges on finalization", {epoch: finalized.epoch}, e as Error); - } + // try { + // const finalizedBlockFC = chain.forkChoice.getBlockHexDefaultStatus(finalized.rootHex); + // if (finalizedBlockFC && finalizedBlockFC.slot > chain.anchorStateLatestBlockSlot) { + // await db.backfilledRanges.put(finalizedBlockFC.slot, chain.anchorStateLatestBlockSlot); + // + // const filteredSeqs = await db.backfilledRanges.entries({ + // gt: chain.anchorStateLatestBlockSlot, + // lt: finalizedBlockFC.slot, + // }); + // logger.debug("updated backfilledRanges", { + // key: finalizedBlockFC.slot, + // value: chain.anchorStateLatestBlockSlot, + // }); + // 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)} + // ); + // } + // } + // } catch (e) { + // logger.error("Error updating backfilledRanges on finalization", {epoch: finalized.epoch}, e as Error); + // } } diff --git a/packages/beacon-node/src/db/beacon.ts b/packages/beacon-node/src/db/beacon.ts index a3107dd4cab2..012b146b5403 100644 --- a/packages/beacon-node/src/db/beacon.ts +++ b/packages/beacon-node/src/db/beacon.ts @@ -2,11 +2,11 @@ import {ChainForkConfig} from "@lodestar/config"; import {Db, LevelDbControllerMetrics, encodeKey} from "@lodestar/db"; import {Bucket} from "./buckets.js"; import {IBeaconDb} from "./interface.js"; +import {BackfillStateRepository} from "./repositories/backfillState.js"; import {CheckpointStateRepository} from "./repositories/checkpointState.js"; import { AttesterSlashingRepository, BLSToExecutionChangeRepository, - BackfilledRanges, BestLightClientUpdateRepository, BlobSidecarsArchiveRepository, BlobSidecarsRepository, @@ -23,6 +23,7 @@ import { SyncCommitteeWitnessRepository, VoluntaryExitRepository, } from "./repositories/index.js"; +import {BackfilledRange} from "./single/backfilledRange.js"; export type BeaconDbModules = { config: ChainForkConfig; @@ -55,7 +56,8 @@ export class BeaconDb implements IBeaconDb { syncCommittee: SyncCommitteeRepository; syncCommitteeWitness: SyncCommitteeWitnessRepository; - backfilledRanges: BackfilledRanges; + backfillState: BackfillStateRepository; + backfilledRange: BackfilledRange; constructor( config: ChainForkConfig, @@ -86,7 +88,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.backfillState = new BackfillStateRepository(config, db); + this.backfilledRange = new BackfilledRange(config, db); } close(): Promise { diff --git a/packages/beacon-node/src/db/buckets.ts b/packages/beacon-node/src/db/buckets.ts index 5dfd7a5968b7..d7ac25db1223 100644 --- a/packages/beacon-node/src/db/buckets.ts +++ b/packages/beacon-node/src/db/buckets.ts @@ -56,7 +56,7 @@ export enum Bucket { // altair_lightClientSyncCommitteeProof = 35, // DEPRECATED on v0.32.0 // index_lightClientInitProof = 36, // DEPRECATED on v0.32.0 - backfilled_ranges = 42, // Backfilled From to To, inclusive of both From, To + backfill_state = 42, // Epoch -> EpochBackfillState // Buckets to support LightClient server v2 lightClient_syncCommitteeWitness = 51, // BlockRoot -> SyncCommitteeWitness diff --git a/packages/beacon-node/src/db/interface.ts b/packages/beacon-node/src/db/interface.ts index 15691ba5944d..39b9249db8fe 100644 --- a/packages/beacon-node/src/db/interface.ts +++ b/packages/beacon-node/src/db/interface.ts @@ -3,7 +3,7 @@ import {CheckpointStateRepository} from "./repositories/checkpointState.js"; import { AttesterSlashingRepository, BLSToExecutionChangeRepository, - BackfilledRanges, + BackfillStateRepository, BestLightClientUpdateRepository, BlobSidecarsArchiveRepository, BlobSidecarsRepository, @@ -20,6 +20,7 @@ import { SyncCommitteeWitnessRepository, VoluntaryExitRepository, } from "./repositories/index.js"; +import {BackfilledRange} from "./single/backfilledRange.js"; /** * The DB service manages the data layer of the beacon chain @@ -57,7 +58,9 @@ export interface IBeaconDb { syncCommittee: SyncCommitteeRepository; syncCommitteeWitness: SyncCommitteeWitnessRepository; - backfilledRanges: BackfilledRanges; + // backfill + backfillState: BackfillStateRepository; + backfilledRange: BackfilledRange; pruneHotDb(): Promise; diff --git a/packages/beacon-node/src/db/repositories/backfillState.ts b/packages/beacon-node/src/db/repositories/backfillState.ts new file mode 100644 index 000000000000..b6d5cd810cab --- /dev/null +++ b/packages/beacon-node/src/db/repositories/backfillState.ts @@ -0,0 +1,36 @@ +import {ContainerType, ListBasicType, OptionalType, ValueOf} from "@chainsafe/ssz"; +import {ChainForkConfig} from "@lodestar/config"; +import {DatabaseController, Repository} from "@lodestar/db"; +import {NUMBER_OF_COLUMNS} from "@lodestar/params"; +import {ssz} from "@lodestar/types"; +import {bytesToInt} from "@lodestar/utils"; +import {Bucket, getBucketNameByValue} from "../buckets.js"; +import {BACKFILLED_RANGE_KEY} from "../single/backfilledRange.js"; + +export const backfillStateWrapperSsz = new ContainerType( + { + hasBlock: ssz.Boolean, + hasBlobs: new OptionalType(ssz.Boolean), + columnIndices: new OptionalType(new ListBasicType(ssz.ColumnIndex, NUMBER_OF_COLUMNS)), + }, + {typeName: "BackfillStateWrapper", jsonCase: "eth2"} +); +export type BackfillStateWrapper = ValueOf; + +export class BackfillStateRepository extends Repository { + constructor(config: ChainForkConfig, db: DatabaseController) { + const bucket = Bucket.backfill_state; + super(config, db, bucket, backfillStateWrapperSsz, getBucketNameByValue(bucket)); + } + + encodeKey(key: number): Uint8Array { + if (key === BACKFILLED_RANGE_KEY) { + throw new Error("Reserved key for backfill range singleton object"); + } + return super.encodeKey(key); + } + + decodeKey(data: Uint8Array): number { + return bytesToInt(super.decodeKey(data) as unknown as Uint8Array, "be"); + } +} diff --git a/packages/beacon-node/src/db/repositories/backfilledRanges.ts b/packages/beacon-node/src/db/repositories/backfilledRanges.ts deleted file mode 100644 index c2958a0e972e..000000000000 --- a/packages/beacon-node/src/db/repositories/backfilledRanges.ts +++ /dev/null @@ -1,29 +0,0 @@ -import {ChainForkConfig} from "@lodestar/config"; -import {DatabaseController, Repository} from "@lodestar/db"; -import {Slot, ssz} from "@lodestar/types"; -import {bytesToInt} from "@lodestar/utils"; -import {Bucket, getBucketNameByValue} from "../buckets.js"; - -/** - * 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 { - constructor(config: ChainForkConfig, db: DatabaseController) { - const bucket = Bucket.backfilled_ranges; - super(config, db, bucket, ssz.Slot, getBucketNameByValue(bucket)); - } - - decodeKey(data: Buffer): number { - return bytesToInt(super.decodeKey(data) as unknown as Uint8Array, "be"); - } - - getId(_value: Slot): number { - throw new Error("Cannot get the db key from slot"); - } -} diff --git a/packages/beacon-node/src/db/repositories/index.ts b/packages/beacon-node/src/db/repositories/index.ts index 1cb06c055d07..c292c939d176 100644 --- a/packages/beacon-node/src/db/repositories/index.ts +++ b/packages/beacon-node/src/db/repositories/index.ts @@ -1,5 +1,5 @@ export {AttesterSlashingRepository} from "./attesterSlashing.js"; -export {BackfilledRanges} from "./backfilledRanges.js"; +export {BackfillStateRepository} from "./backfillState.js"; export {BlobSidecarsRepository} from "./blobSidecars.js"; export {BlobSidecarsArchiveRepository} from "./blobSidecarsArchive.js"; export {BlockRepository} from "./block.js"; diff --git a/packages/beacon-node/src/db/single/backfilledRange.ts b/packages/beacon-node/src/db/single/backfilledRange.ts new file mode 100644 index 000000000000..babe9b4d34c2 --- /dev/null +++ b/packages/beacon-node/src/db/single/backfilledRange.ts @@ -0,0 +1,37 @@ +import {ContainerType, ValueOf} from "@chainsafe/ssz"; +import {ChainForkConfig} from "@lodestar/config"; +import {Db, DbReqOpts, encodeKey} from "@lodestar/db"; +import {ssz} from "@lodestar/types"; +import {Bucket, getBucketNameByValue} from "../buckets.js"; + +export const backfilledRangeWrapperSsz = new ContainerType( + {beginningEpoch: ssz.Epoch, endingEpoch: ssz.Epoch}, + {typeName: "BackfilledRange", jsonCase: "eth2"} +); +export type BackfilledRangeWrapper = ValueOf; +export const BACKFILLED_RANGE_KEY = -1; + +export class BackfilledRange { + private readonly db: Db; + private readonly key: Uint8Array; + private readonly dbReqOpts: DbReqOpts; + + constructor(_config: ChainForkConfig, db: Db) { + this.db = db; + this.key = encodeKey(Bucket.backfill_state, BACKFILLED_RANGE_KEY); + this.dbReqOpts = {bucketId: getBucketNameByValue(Bucket.backfill_state)}; + } + + async put(value: BackfilledRangeWrapper): Promise { + await this.db.put(this.key, backfilledRangeWrapperSsz.serialize(value), this.dbReqOpts); + } + + async get(): Promise { + const value = await this.db.get(this.key, this.dbReqOpts); + return value ? backfilledRangeWrapperSsz.deserialize(value) : null; + } + + async delete(): Promise { + await this.db.delete(this.key, this.dbReqOpts); + } +} diff --git a/packages/beacon-node/src/db/single/index.ts b/packages/beacon-node/src/db/single/index.ts new file mode 100644 index 000000000000..0192467d1dc6 --- /dev/null +++ b/packages/beacon-node/src/db/single/index.ts @@ -0,0 +1 @@ +export {BACKFILLED_RANGE_KEY, BackfilledRange} from "./backfilledRange.js"; diff --git a/packages/beacon-node/src/sync/backfill/backfill.ts b/packages/beacon-node/src/sync/backfill/backfill.ts index 2e2f4eb0df65..cc7192c6f850 100644 --- a/packages/beacon-node/src/sync/backfill/backfill.ts +++ b/packages/beacon-node/src/sync/backfill/backfill.ts @@ -240,13 +240,10 @@ export class BackfillSync extends (EventEmitter as {new (): BackfillSyncEmitter} lastBackSyncedBlock: null, }; - // Load the previous written to slot for the key backfillStartFromSlot - // in backfilledRanges + // TODO: migrate to backfillState const backfillStartFromSlot = anchorSlot; - const backfillRangeWrittenSlot = await db.backfilledRanges.get(backfillStartFromSlot); - const previousBackfilledRanges = await db.backfilledRanges.entries({ - lte: backfillStartFromSlot, - }); + const backfillRangeWrittenSlot = null as number | null; + const previousBackfilledRanges = [] as {key: number; value: number}[]; modules.logger.info("Initializing from Checkpoint", { root: toRootHex(anchorCp.root), epoch: anchorCp.epoch, @@ -393,7 +390,8 @@ export class BackfillSync extends (EventEmitter as {new (): BackfillSyncEmitter} this.syncAnchor.lastBackSyncedBlock.slot < this.backfillRangeWrittenSlot ) { this.backfillRangeWrittenSlot = this.syncAnchor.lastBackSyncedBlock.slot; - await this.db.backfilledRanges.put(this.backfillStartFromSlot, this.backfillRangeWrittenSlot); + // TODO: migrate to backfillState + // await this.db.backfilledRanges.put(this.backfillStartFromSlot, this.backfillRangeWrittenSlot); this.logger.debug( `Updated the backfill range from=${this.backfillStartFromSlot} till=${this.backfillRangeWrittenSlot}` ); @@ -547,105 +545,97 @@ export class BackfillSync extends (EventEmitter as {new (): BackfillSyncEmitter} throw Error("Backfill ranges can only be used once we have a valid lastBackSyncedBlock as a pivot point"); } - let validSequence = false; + const validSequence = false; if (this.syncAnchor.lastBackSyncedBlock.slot === null) return validSequence; const lastBackSyncedSlot = this.syncAnchor.lastBackSyncedBlock.slot; - const filteredSeqs = await this.db.backfilledRanges.entries({ - gte: lastBackSyncedSlot, - }); - - if (filteredSeqs.length > 0) { - const jumpBackTo = Math.min(...filteredSeqs.map(({value: justToSlot}) => justToSlot)); - - if (jumpBackTo < lastBackSyncedSlot) { - validSequence = true; - const anchorBlock = await this.db.blockArchive.get(jumpBackTo); - if (!anchorBlock) { - validSequence = false; - this.logger.warn( - `Invalid backfill sequence: expected a block at ${jumpBackTo} in blockArchive, ignoring the sequence` - ); - } - if (anchorBlock && validSequence && this.prevFinalizedCheckpointBlock.slot >= jumpBackTo) { - this.logger.debug( - `Found a sequence going back to ${jumpBackTo} before the previous finalized or wsCheckpoint`, - {slot: this.prevFinalizedCheckpointBlock.slot} - ); - - // Everything saved in db between a backfilled range is a connected sequence - // we only need to check if prevFinalizedCheckpointBlock is in db - const prevBackfillCpBlock = await this.db.blockArchive.getByRoot(this.prevFinalizedCheckpointBlock.root); - if ( - prevBackfillCpBlock != null && - this.prevFinalizedCheckpointBlock.slot === prevBackfillCpBlock.message.slot - ) { - this.logger.verbose("Validated current prevFinalizedCheckpointBlock", { - root: toRootHex(this.prevFinalizedCheckpointBlock.root), - slot: prevBackfillCpBlock.message.slot, - }); - } else { - validSequence = false; - this.logger.warn( - `Invalid backfill sequence: previous finalized or checkpoint block root=${toRootHex( - this.prevFinalizedCheckpointBlock.root - )}, slot=${this.prevFinalizedCheckpointBlock.slot} ${ - prevBackfillCpBlock ? "found at slot=" + prevBackfillCpBlock.message.slot : "not found" - }, ignoring the sequence` - ); - } - } - - if (anchorBlock && validSequence) { - // Update the current sequence in DB as we will be cleaning up previous sequences - await this.db.backfilledRanges.put(this.backfillStartFromSlot, jumpBackTo); - this.backfillRangeWrittenSlot = jumpBackTo; - this.logger.verbose( - `Jumped and updated the backfilled range ${this.backfillStartFromSlot}, ${this.backfillRangeWrittenSlot}`, - {jumpBackTo} - ); - - const anchorBlockHeader = blockToHeader(this.config, anchorBlock.message); - const anchorBlockRoot = ssz.phase0.BeaconBlockHeader.hashTreeRoot(anchorBlockHeader); - - this.syncAnchor = { - anchorBlock, - anchorBlockRoot, - anchorSlot: jumpBackTo, - lastBackSyncedBlock: {root: anchorBlockRoot, slot: jumpBackTo, block: anchorBlock}, - }; - if (this.prevFinalizedCheckpointBlock.slot >= jumpBackTo) { - // prevFinalizedCheckpointBlock must have been validated, update to a - // new unverified - // finalized or wsCheckpoint behind the new lastBackSyncedBlock - this.prevFinalizedCheckpointBlock = await extractPreviousFinOrWsCheckpoint( - this.config, - this.db, - jumpBackTo, - this.logger - ); - } - - this.metrics?.backfillSync.totalBlocks.inc( - {method: BackfillSyncMethod.backfilled_ranges}, - lastBackSyncedSlot - jumpBackTo - ); - } - } - } - - // Only delete < backfillStartFromSlot, the keys greater than this would be cleaned - // up by the archival process of forward sync - const cleanupSeqs = filteredSeqs.filter((entry) => entry.key < this.backfillStartFromSlot); - if (cleanupSeqs.length > 0) { - await this.db.backfilledRanges.batchDelete(cleanupSeqs.map((entry) => entry.key)); - this.logger.debug( - `Cleaned up the old sequences between ${this.backfillStartFromSlot},${toRootHex( - this.syncAnchor.lastBackSyncedBlock.root - )}`, - {cleanupSeqs: JSON.stringify(cleanupSeqs)} - ); - } + // const filteredSeqs = await this.db.backfilledRanges.entries({ + // gte: lastBackSyncedSlot, + // }); + // + // if (filteredSeqs.length > 0) { + // const jumpBackTo = Math.min(...filteredSeqs.map(({value: justToSlot}) => justToSlot)); + // + // if (jumpBackTo < lastBackSyncedSlot) { + // validSequence = true; + // const anchorBlock = await this.db.blockArchive.get(jumpBackTo); + // if (!anchorBlock) { + // validSequence = false; + // this.logger.warn( + // `Invalid backfill sequence: expected a block at ${jumpBackTo} in blockArchive, ignoring the sequence` + // ); + // } + // if (anchorBlock && validSequence && this.prevFinalizedCheckpointBlock.slot >= jumpBackTo) { + // this.logger.debug( + // `Found a sequence going back to ${jumpBackTo} before the previous finalized or wsCheckpoint`, + // {slot: this.prevFinalizedCheckpointBlock.slot} + // ); + // + // const prevBackfillCpBlock = await this.db.blockArchive.getByRoot(this.prevFinalizedCheckpointBlock.root); + // if ( + // prevBackfillCpBlock != null && + // this.prevFinalizedCheckpointBlock.slot === prevBackfillCpBlock.message.slot + // ) { + // this.logger.verbose("Validated current prevFinalizedCheckpointBlock", { + // root: toRootHex(this.prevFinalizedCheckpointBlock.root), + // slot: prevBackfillCpBlock.message.slot, + // }); + // } else { + // validSequence = false; + // this.logger.warn( + // `Invalid backfill sequence: previous finalized or checkpoint block root=${toRootHex( + // this.prevFinalizedCheckpointBlock.root + // )}, slot=${this.prevFinalizedCheckpointBlock.slot} ${ + // prevBackfillCpBlock ? "found at slot=" + prevBackfillCpBlock.message.slot : "not found" + // }, ignoring the sequence` + // ); + // } + // } + // + // if (anchorBlock && validSequence) { + // await this.db.backfilledRanges.put(this.backfillStartFromSlot, jumpBackTo); + // this.backfillRangeWrittenSlot = jumpBackTo; + // this.logger.verbose( + // `Jumped and updated the backfilled range ${this.backfillStartFromSlot}, ${this.backfillRangeWrittenSlot}`, + // {jumpBackTo} + // ); + // + // const anchorBlockHeader = blockToHeader(this.config, anchorBlock.message); + // const anchorBlockRoot = ssz.phase0.BeaconBlockHeader.hashTreeRoot(anchorBlockHeader); + // + // this.syncAnchor = { + // anchorBlock, + // anchorBlockRoot, + // anchorSlot: jumpBackTo, + // lastBackSyncedBlock: {root: anchorBlockRoot, slot: jumpBackTo, block: anchorBlock}, + // }; + // if (this.prevFinalizedCheckpointBlock.slot >= jumpBackTo) { + // this.prevFinalizedCheckpointBlock = await extractPreviousFinOrWsCheckpoint( + // this.config, + // this.db, + // jumpBackTo, + // this.logger + // ); + // } + // + // this.metrics?.backfillSync.totalBlocks.inc( + // {method: BackfillSyncMethod.backfilled_ranges}, + // lastBackSyncedSlot - jumpBackTo + // ); + // } + // } + // } + // + // const cleanupSeqs = filteredSeqs.filter((entry) => entry.key < this.backfillStartFromSlot); + // if (cleanupSeqs.length > 0) { + // await this.db.backfilledRanges.batchDelete(cleanupSeqs.map((entry) => entry.key)); + // this.logger.debug( + // `Cleaned up the old sequences between ${this.backfillStartFromSlot},${toRootHex( + // this.syncAnchor.lastBackSyncedBlock.root + // )}`, + // {cleanupSeqs: JSON.stringify(cleanupSeqs)} + // ); + // } return validSequence; } diff --git a/packages/beacon-node/src/sync/backfill/backfillV2.ts b/packages/beacon-node/src/sync/backfill/backfillV2.ts new file mode 100644 index 000000000000..2baa548e72de --- /dev/null +++ b/packages/beacon-node/src/sync/backfill/backfillV2.ts @@ -0,0 +1,754 @@ +import {EventEmitter} from "node:events"; +import {PeerId} from "@libp2p/interface"; +import {StrictEventEmitter} from "strict-event-emitter-types"; +import {BeaconConfig} from "@lodestar/config"; +import {SLOTS_PER_EPOCH, SLOTS_PER_HISTORICAL_ROOT} from "@lodestar/params"; +import {IBeaconStateView, computeEpochAtSlot, computeStartSlotAtEpoch} from "@lodestar/state-transition"; +import {Epoch, Root, RootHex, SignedBeaconBlock, Slot, ssz} from "@lodestar/types"; +import {ErrorAborted, Logger, sleep, toRootHex} from "@lodestar/utils"; +import {IBeaconChain} from "../../chain/index.js"; +import {GENESIS_SLOT, ZERO_HASH} from "../../constants/index.js"; +import {IBeaconDb} from "../../db/index.js"; +import {Metrics} from "../../metrics/metrics.js"; +import {INetwork, NetworkEvent, NetworkEventData, PeerAction} from "../../network/index.js"; +import {ItTrigger} from "../../util/itTrigger.js"; +import {PeerIdStr} from "../../util/peerId.js"; +import {shuffle} from "../../util/shuffle.js"; +import {BackfillSyncMethod} from "./backfill.js"; +import {BackfillSyncError, BackfillSyncErrorCode} from "./errors.js"; +import {verifyBlockProposerSignature} from "./verify.js"; + +const EPOCH_FLUSH_YIELD_MS = 50; //TODO: remove this since we will change to separate thread for backfill sync. +const MAX_PEER_FAILURES = 5; +const DEFAULT_BACKFILL_BATCH_SIZE = 64; + +// Phase 2 hedged fetch: a single byRoot request is raced across up to this many peers. +const HEDGE_PEER_COUNT = 3; +const HEDGE_DELAY_MS = 400; + +export type BackfillSyncModules = { + chain: IBeaconChain; + db: IBeaconDb; + network: INetwork; + config: BeaconConfig; + logger: Logger; + metrics: Metrics | null; + anchorState: IBeaconStateView; + signal: AbortSignal; +}; + +export type BackfillSyncOpts = { + backfillBatchSize: number; + backfillToGenesis: boolean; +}; + +export enum BackfillSyncEvent { + completed = "BackfillSync-completed", +} + +export enum BackfillSyncStatus { + pending = "pending", + syncing = "syncing", + completed = "completed", + aborted = "aborted", +} + +// numbers to stored in Prometheus metrics +const syncStatus: {[K in BackfillSyncStatus]: number} = { + [BackfillSyncStatus.aborted]: 0, + [BackfillSyncStatus.pending]: 1, + [BackfillSyncStatus.syncing]: 2, + [BackfillSyncStatus.completed]: 3, +}; + +type BackfillSyncEvents = { + [BackfillSyncEvent.completed]: (oldestSlotSynced: Slot) => void; +}; + +type BackfillSyncEmitter = StrictEventEmitter; + +// per peer metadata; we should record this to recognize bad peer +type PeerMeta = { + failedRequests: number; +}; + +// A block obtained from a specific peer during a Phase 2 hedged fetch. +type RacedBlock = {block: SignedBeaconBlock; peer: PeerIdStr}; + +/** + * BackfillSync walks the chain backwards from the checkpoint-sync anchor down to genesis, + * fetching every block via `BeaconBlocksByRoot`. It runs in two phases: + * + * Phase 1 — batched byRoot (the `block_roots` window): + * A `BeaconState` carries `block_roots`, a circular buffer of the canonical block roots + * for the most recent `SLOTS_PER_HISTORICAL_ROOT` (8192) slots. The anchor state therefore + * already knows the roots for that whole window. Therefore, we could request them in large + * batches and verify them with `hashTreeRoot(block) === knownRoot` instead of checking the + * chain linkage. + * + * Phase 2 — serial parentRoot walk (older than the window): + * Below the `block_roots` window the anchor state no longer knows the roots, so we walk + * one block at a time, using each block's `parentRoot` as the next root to request. The + * walk is inherently serial, so each single byRoot request is hedged across several peers + * (`fetchBlockByRootRaced`) to hide per-request failure and tail latency. + */ +export class BackfillSync extends (EventEmitter as {new (): BackfillSyncEmitter}) { + private status: BackfillSyncStatus = BackfillSyncStatus.pending; + private anchorRoot: Root; + private anchorSlot: Slot; + + private epochBuffer: SignedBeaconBlock[] = []; + private knownRoots: Root[] = []; // roots from phase 1 + + private readonly chain: IBeaconChain; + private readonly db: IBeaconDb; + private readonly network: INetwork; + private readonly config: BeaconConfig; + private readonly logger: Logger; + private readonly metrics: Metrics | null; + private readonly anchorState: IBeaconStateView; + + private readonly batchSize: number; + // Oldest slot we will backfill down to. `null` when backfilling all the way to genesis + // (in which case the loop terminates via the `anchorRoot === ZERO_HASH` check instead). + private readonly stopSlot: Slot | null; + + private processor = new ItTrigger(); + private peers = new Map(); + // Peer that last served a block; `pickPeers` lists it first so a hedged fetch + // reuses an already-warm connection instead of dialing a fresh random peer. + private lastGoodPeer: PeerIdStr | null = null; + private signal: AbortSignal; + + constructor(opts: BackfillSyncOpts, modules: BackfillSyncModules, anchorRoot: Root, anchorSlot: Slot) { + super(); + + this.anchorRoot = anchorRoot; + this.anchorSlot = anchorSlot; + + this.chain = modules.chain; + this.db = modules.db; + this.network = modules.network; + this.config = modules.config; + this.logger = modules.logger; + this.metrics = modules.metrics; + this.anchorState = modules.anchorState; + this.processor = new ItTrigger(); + this.peers = new Map(); + this.signal = modules.signal; + + // Number of roots requested per batched byRoot call in Phase 1. + // Capped by `MAX_REQUEST_BLOCKS_DENEB` + this.batchSize = Math.max( + 1, + Math.min(opts.backfillBatchSize || DEFAULT_BACKFILL_BATCH_SIZE, this.config.MAX_REQUEST_BLOCKS_DENEB) + ); + + // `getLatestWeakSubjectivityCheckpointEpoch()` returns `currentEpoch - wsPeriod`; clamp + // to GENESIS_SLOT for anchors younger than one WS period (e.g. fresh testnets). + if (opts.backfillToGenesis) { + this.stopSlot = null; + } else { + const wsEpoch = Math.max(0, this.anchorState.getLatestWeakSubjectivityCheckpointEpoch()); + this.stopSlot = computeStartSlotAtEpoch(wsEpoch); + } + + this.network.events.on(NetworkEvent.peerConnected, this.addPeer); + this.network.events.on(NetworkEvent.peerDisconnected, this.removePeer); + + this.sync() + .then((oldestSlotSynced) => { + if (this.status !== BackfillSyncStatus.completed) { + throw new ErrorAborted(`Invalid BackfillSyncStatus at completion: status = ${this.status}`); + } + this.emit(BackfillSyncEvent.completed, oldestSlotSynced); + this.logger.info("BackfillSync completed", {oldestSlotSynced}); + this.close(); + }) + .catch((e) => { + if (!(e instanceof ErrorAborted)) { + this.logger.error("BackfillSync processor error", e); + } + this.status = BackfillSyncStatus.aborted; + this.close(); + }); + + const metrics = this.metrics; + if (metrics) { + metrics.backfillSync.status.addCollect(() => metrics.backfillSync.status.set(syncStatus[this.status])); + metrics.backfillSync.backfilledTillSlot.addCollect(() => + metrics.backfillSync.backfilledTillSlot.set(this.anchorSlot) + ); + } + } + + /** + * Initialize backfill sync from anchor state and DB. + * Checks BackfilledRange to resume from a previous session. + */ + static async init(opts: BackfillSyncOpts, modules: BackfillSyncModules): Promise { + const {anchorState, db, logger} = modules; + + const {checkpoint: anchorCp} = anchorState.computeAnchorCheckpoint(); + const anchorSlot = anchorState.latestBlockHeader.slot; + const anchorEpoch = computeEpochAtSlot(anchorSlot); + + const backfilledRange = await db.backfilledRange.get(); + let startRoot: Root; + let startSlot: Slot; + + if (backfilledRange && backfilledRange.beginningEpoch === anchorEpoch) { + const resumeEpoch = backfilledRange.endingEpoch; + const resumeSlot = resumeEpoch * SLOTS_PER_EPOCH; + + const boundaryBlock = await db.blockArchive.get(resumeSlot); + if (boundaryBlock) { + startRoot = boundaryBlock.message.parentRoot; + startSlot = resumeSlot; + logger.info("Resuming backfill sync from previous session", { + resumeEpoch, + resumeSlot, + beginningEpoch: backfilledRange.beginningEpoch, + }); + } else { + startRoot = anchorCp.root; + startSlot = anchorSlot; + logger.warn("BackfilledRange exists but boundary block missing, starting fresh", { + resumeEpoch, + }); + } + } else { + startRoot = anchorCp.root; + startSlot = anchorSlot; + + await db.backfilledRange.put({ + beginningEpoch: anchorEpoch, + endingEpoch: 0, + }); + + logger.info("Starting fresh backfill sync", { + anchorSlot, + anchorEpoch, + anchorRoot: toRootHex(anchorCp.root), + }); + } + return new BackfillSync(opts, modules, startRoot, startSlot); + } + + private async sync(): Promise { + await this.buildKnownRoots(); + this.processor.trigger(); + + for await (const _ of this.processor) { + if (this.status === BackfillSyncStatus.aborted) break; + + // Reached the chain root: anchorRoot becomes ZERO_HASH only as the parentRoot of + // the genesis block, so there is nothing left to fetch — flush and complete. + if (ssz.Root.equals(this.anchorRoot, ZERO_HASH)) { + if (this.epochBuffer.length > 0) { + const bufferEpoch = computeEpochAtSlot(this.epochBuffer[0].message.slot); + await this.flushEpoch(bufferEpoch); + } + this.status = BackfillSyncStatus.completed; + return this.anchorSlot; + } + + // Reached the WS period floor: enough history for weak subjectivity, no need to go further. + if (this.stopSlot !== null && this.anchorSlot <= this.stopSlot) { + if (this.epochBuffer.length > 0) { + const bufferEpoch = computeEpochAtSlot(this.epochBuffer[0].message.slot); + await this.flushEpoch(bufferEpoch); + } + this.logger.info("BackfillSync reached WS period floor", { + stopSlot: this.stopSlot, + anchorSlot: this.anchorSlot, + }); + this.status = BackfillSyncStatus.completed; + return this.anchorSlot; + } + + this.status = BackfillSyncStatus.syncing; + + // Phase 1 walks the known `block_roots` window one batch at a time; Phase 2 walks + // older blocks one parentRoot at a time, hedged across several peers. + const isPhase1 = this.knownRoots.length > 0; + const peers = this.pickPeers(isPhase1 ? 1 : HEDGE_PEER_COUNT); + if (peers.length === 0) { + this.status = BackfillSyncStatus.pending; + this.logger.debug("BackfillSync: no eligible peers, waiting"); + continue; + } + try { + if (isPhase1) { + await this.syncBatchByRoot(peers[0]); + } else { + await this.syncOneByParentRoot(peers); + } + } catch (e) { + // Per-peer accounting (penalize / reportPeer) is done by the sync method that threw; + // the loop only logs and handles a fatal abort. + if (e instanceof BackfillSyncError) { + switch (e.type.code) { + case BackfillSyncErrorCode.INTERNAL_ERROR: + this.status = BackfillSyncStatus.aborted; + this.logger.error("BackfillSync error", {}, e); + break; + + case BackfillSyncErrorCode.MISSING_BLOCK: + // No peer served the block this round; the loop retries with a fresh peer set. + this.logger.debug("BackfillSync could not fetch block, will retry", {code: e.type.code}); + break; + + default: + this.logger.warn("BackfillSync peer request failed", {code: e.type.code}); + } + } else { + this.metrics?.backfillSync.errors.inc(); + this.logger.error("BackfillSync error", {}, e as Error); + } + + if (this.status === BackfillSyncStatus.aborted) break; + } + if (this.signal.aborted) break; + this.processor.trigger(); + } + + throw new ErrorAborted("BackfillSync"); + } + + /** + * Phase 1 setup: enumerate the canonical block roots the anchor state already knows. + * + * `block_roots` covers slots [state.slot - SLOTS_PER_HISTORICAL_ROOT, state.slot). For each + * slot in that window `getBlockRootAtSlot` returns the root of the most recent block at or + * before it, so skipped slots repeat the previous root — we dedupe consecutive equal roots + * to recover exactly one entry per real block, ordered from `anchorSlot` down to the floor. + * + * If the anchor state cannot supply the window (e.g. the resume point already sits below it) + * `knownRoots` is left empty and the sync runs as a pure Phase 2 serial walk. + */ + private async buildKnownRoots(): Promise { + try { + const floorSlot = Math.max(GENESIS_SLOT, this.anchorState.slot - SLOTS_PER_HISTORICAL_ROOT); + const roots: Root[] = []; + let prevHex: RootHex | null = null; + + for (let slot = this.anchorSlot; slot >= floorSlot; slot--) { + // `getBlockRootAtSlot` throws for slot >= state.slot; the anchor root is already known. + const root = slot === this.anchorSlot ? this.anchorRoot : this.anchorState.getBlockRootAtSlot(slot); + const hex = toRootHex(root); + if (hex !== prevHex) { + roots.push(root); + prevHex = hex; + } + } + + // Drop roots whose blocks are already in `blockArchive` (e.g. from an earlier run with + // an overlapping window). `getSlotByRoot` only reads the small root→slot index entry, + // not the whole block, so the pre-filter is cheap even when nothing overlaps. + const presentSlots = await Promise.all(roots.map((r) => this.db.blockArchive.getSlotByRoot(r))); + const filtered = roots.filter((_, i) => presentSlots[i] === null); + const skippedAlreadyStored = roots.length - filtered.length; + + this.knownRoots = filtered; + this.logger.info("BackfillSync: built known roots window", { + knownRoots: filtered.length, + skippedAlreadyStored, + fromSlot: this.anchorSlot, + floorSlot, + }); + } catch (e) { + // Defensive: degrade to a pure serial parentRoot walk rather than aborting the sync. + this.knownRoots = []; + this.logger.warn("BackfillSync: could not build known roots window, falling back to serial walk", {}, e as Error); + } + } + + /** + * Phase 1 step: request a batch of known roots in a single byRoot call. + * + * The roots come from the trusted anchor state, so verification is just + * `hashTreeRoot(block) === requestedRoot` plus a (batched) proposer-signature check — no + * chain-linkage check is needed. A peer may omit roots it does not have; we consume only + * the contiguous prefix it returned and leave the rest queued for another peer. + */ + private async syncBatchByRoot(peer: PeerIdStr): Promise { + const batch = this.knownRoots.slice(0, this.batchSize); + const blocks = await this.network.sendBeaconBlocksByRoot(peer, batch); + + const blockByRoot = new Map(); + for (const block of blocks) { + const root = this.config.getForkTypes(block.message.slot).BeaconBlock.hashTreeRoot(block.message); + blockByRoot.set(toRootHex(root), block); + } + + // Take the contiguous prefix of requested roots the peer actually returned, so the + // walk stays gap-free; any trailing missing roots remain queued for retry. + const received: SignedBeaconBlock[] = []; + for (const root of batch) { + const block = blockByRoot.get(toRootHex(root)); + if (!block) break; + received.push(block); + } + + if (received.length === 0) { + this.onPeerRequestFailed(peer); + this.metrics?.backfillSync.errors.inc(); + throw new BackfillSyncError({ + code: BackfillSyncErrorCode.MISSING_BLOCK, + root: batch[0], + peerId: peer as unknown as PeerId, + }); + } + + try { + await verifyBlockProposerSignature(this.config, this.chain.bls, received); + } catch (e) { + this.network.reportPeer(peer, PeerAction.LowToleranceError, "BackfillSyncInvalidSignature"); + this.onPeerRequestFailed(peer); + this.metrics?.backfillSync.errors.inc(); + throw e; + } + + // `received` is ordered highest-slot first (knownRoots is anchor-down), matching the + // backward-walk order the epoch buffer expects. + for (const block of received) { + await this.ingestBlock(block); + } + + this.knownRoots.splice(0, received.length); + this.onPeerRequestSuccess(peer); + this.lastGoodPeer = peer; + this.metrics?.backfillSync.totalBlocks.inc({method: BackfillSyncMethod.blockbyroot}, received.length); + + if (this.knownRoots.length === 0) { + // Window exhausted — hand off to the Phase 2 serial walk via the oldest block's parent. + // biome-ignore lint/style/noNonNullAssertion: received is non-empty (checked above) + const oldest = received.at(-1)!; + this.anchorRoot = oldest.message.parentRoot; + this.anchorSlot = oldest.message.slot; + } else if (received.length < batch.length) { + // Peer was missing part of the batch; the remaining roots stay queued for another peer. + this.onPeerRequestFailed(peer); + } + } + + /** + * Phase 2 step: fetch the next block by its root and walk to its parent. + * + * The walk is inherently serial — block N must arrive before N-1's root is known — so the + * single byRoot request is hedged across several peers to hide per-request failure and + * tail latency (see `fetchBlockByRootRaced`). + */ + private async syncOneByParentRoot(peers: PeerIdStr[]): Promise { + // Cache short-circuit: if this block is already in `blockArchive` (overlap with an + // earlier run, or a contiguous already-filled segment), use the cached copy — no network + // fetch, no signature recheck, no epoch-buffer churn. Just walk to its parent. + const cached = await this.db.blockArchive.getByRoot(this.anchorRoot); + if (cached !== null) { + this.anchorSlot = cached.message.slot; + this.anchorRoot = cached.message.parentRoot; + this.metrics?.backfillSync.totalBlocks.inc({method: BackfillSyncMethod.database}, 1); + + // Still run the epoch-boundary lookahead so we can jump whole already-filled epochs + // in one bulk skip instead of per-block walking through 32+ cached blocks per epoch. + if (cached.message.slot !== GENESIS_SLOT) { + const blockEpoch = computeEpochAtSlot(cached.message.slot); + const nextEpoch = computeEpochAtSlot(cached.message.slot - 1); + if (nextEpoch !== blockEpoch) { + const existingState = await this.db.backfillState.get(nextEpoch); + if (existingState?.hasBlock) { + this.logger.info("Skipping already-filled epoch", {epoch: nextEpoch}); + await this.skipFilledEpochs(nextEpoch); + } + } + } + return; + } + + const result = await this.fetchBlockByRootRaced(this.anchorRoot, peers); + + if (result === null) { + // None of the hedged peers served the block; the loop retries with a fresh peer set + // (peers that keep failing are eventually excluded by `pickPeers`). + throw new BackfillSyncError({ + code: BackfillSyncErrorCode.MISSING_BLOCK, + root: this.anchorRoot, + peerId: peers[0] as unknown as PeerId, + }); + } + + const {block, peer} = result; + + // `fetchBlockByRootRaced` already verified hashTreeRoot(block) === anchorRoot, so the + // block body is canonical. The signature is separate data the peer could still have + // tampered with, so it is verified here. + if (block.message.slot !== GENESIS_SLOT) { + try { + await verifyBlockProposerSignature(this.config, this.chain.bls, [block]); + } catch (e) { + this.network.reportPeer(peer, PeerAction.LowToleranceError, "BackfillSyncInvalidSignature"); + this.onPeerRequestFailed(peer); + this.metrics?.backfillSync.errors.inc(); + throw e; + } + } + + await this.ingestBlock(block); + this.anchorRoot = block.message.parentRoot; + this.onPeerRequestSuccess(peer); + this.lastGoodPeer = peer; + this.metrics?.backfillSync.totalBlocks.inc({method: BackfillSyncMethod.blockbyroot}, 1); + + // After genesis, parentRoot is ZERO_HASH and the loop's top-of-iteration + // check handles flush+complete; skip the already-filled lookahead here. + // TODO: we're not syncing to GENESIS unless the user wants to do so. + if (block.message.slot !== GENESIS_SLOT) { + const blockEpoch = computeEpochAtSlot(block.message.slot); + const nextEpoch = computeEpochAtSlot(block.message.slot - 1); + if (nextEpoch !== blockEpoch) { + const existingState = await this.db.backfillState.get(nextEpoch); + if (existingState?.hasBlock) { + this.logger.info("Skipping already-filled epoch", {epoch: nextEpoch}); + await this.skipFilledEpochs(nextEpoch); + } + } + } + } + + /** + * Hedged byRoot fetch for a single block. + * + * The request goes to the primary peer first; the remaining peers are contacted only if + * the primary has not produced the block within `HEDGE_DELAY_MS` (or fails outright). + * Delayed hedging keeps the fast common case at 1x bandwidth — extra peers pay off only on + * the slow/failing tail. Returns the first peer whose block matches `root` exactly, or + * `null` if none of the peers could serve it. + * + * Note: the network API exposes no AbortSignal, so a losing in-flight request cannot be + * truly cancelled — it completes and its result is discarded. Delayed hedging bounds that + * waste to the tail. + */ + private async fetchBlockByRootRaced(root: Root, peers: PeerIdStr[]): Promise { + const rootHex = toRootHex(root); + const primary = this.attemptFetch(root, rootHex, peers[0]); + const rest = peers.slice(1); + + if (rest.length === 0) { + return primary; + } + + // Wait for the primary peer, but no longer than the hedge delay. + let hedgeFired = false; + const primaryResult = await Promise.race([ + primary, + sleep(HEDGE_DELAY_MS).then((): RacedBlock | null => { + hedgeFired = true; + return null; + }), + ]); + + // Primary served the block before the hedge delay — no fan-out needed. + if (!hedgeFired && primaryResult !== null) { + return primaryResult; + } + + // Primary was too slow or failed: fan out to the remaining peers, keeping the primary + // in the race if it timed out (still in flight rather than already-failed). + const racers = rest.map((peer) => this.attemptFetch(root, rootHex, peer)); + if (hedgeFired) { + racers.push(primary); + } + return this.firstSuccessful(racers); + } + + /** + * Request one block by root from a single peer. Never rejects: a reqresp failure, a + * missing block, or a non-matching block all resolve to `null` and mark the peer failed. + * A non-null result is guaranteed to match `root` exactly. + */ + private async attemptFetch(root: Root, rootHex: RootHex, peer: PeerIdStr): Promise { + try { + const [block] = await this.network.sendBeaconBlocksByRoot(peer, [root]); + if (block) { + const blockRoot = this.config.getForkTypes(block.message.slot).BeaconBlock.hashTreeRoot(block.message); + if (toRootHex(blockRoot) === rootHex) { + return {block, peer}; + } + } + } catch { + // reqresp-level failure (dial error, timeout, server error, ...) — treated as a miss + } + this.onPeerRequestFailed(peer); + this.metrics?.backfillSync.errors.inc(); + return null; + } + + /** + * Resolve with the first promise that yields a non-null value, or `null` once every + * promise has resolved to `null`. Input promises never reject (see `attemptFetch`). + */ + private firstSuccessful(attempts: Promise[]): Promise { + return new Promise((resolve) => { + let pending = attempts.length; + for (const attempt of attempts) { + void attempt.then((result) => { + if (result !== null) { + resolve(result); + } else if (--pending === 0) { + resolve(null); + } + }); + } + }); + } + + /** + * Append a block to the epoch buffer while walking backward, flushing the previous + * (higher) epoch once a block from a lower epoch arrives. + * + * Comparing against the buffer's epoch (rather than epochAt(slot-1)) handles skipped + * slots at epoch boundaries — e.g. walk 97→95 with slot 96 missing. + */ + private async ingestBlock(block: SignedBeaconBlock): Promise { + const blockEpoch = computeEpochAtSlot(block.message.slot); + + if (this.epochBuffer.length > 0) { + const bufferEpoch = computeEpochAtSlot(this.epochBuffer[0].message.slot); + if (bufferEpoch !== blockEpoch) { + await this.flushEpoch(bufferEpoch); + } + } + + this.epochBuffer.push(block); + this.anchorSlot = block.message.slot; + } + + // storing buffered blocks, update the backfill range. + private async flushEpoch(epoch: Epoch): Promise { + if (this.epochBuffer.length === 0) return; + + const puts = this.epochBuffer.map((block) => ({ + key: block.message.slot, + value: block, + })); + await this.db.blockArchive.batchPut(puts); + + await this.db.backfillState.put(epoch, { + hasBlock: true, + hasBlobs: null, // TODO: add blob backfills in the future + columnIndices: null, + }); + + const range = await this.db.backfilledRange.get(); + if (range) { + await this.db.backfilledRange.put({ + beginningEpoch: range.beginningEpoch, + endingEpoch: epoch, + }); + } + + this.logger.verbose("Flushed epoch to DB", { + epoch, + blocks: this.epochBuffer.length, + }); + + this.epochBuffer = []; + + await sleep(EPOCH_FLUSH_YIELD_MS, this.signal); + } + + // finds where to resume after skipping over the filled epochs. + // | epoch | 0 .. 4 | 5..6 | 7 | 8 .. 9 | 10 | + // | | xfilled | filled | filled | gap | anchor | + // skipFilledEpochs(7) => sync resumes from epoch 4 + private async skipFilledEpochs(startEpoch: Epoch): Promise { + let epoch = startEpoch; + + while (epoch > 0) { + const state = await this.db.backfillState.get(epoch); + if (!state?.hasBlock) break; + epoch--; + } + + if (epoch < startEpoch) { + const bottomSlot = (epoch + 1) * SLOTS_PER_EPOCH; + const blocks = await this.db.blockArchive.values({ + gte: bottomSlot, + limit: 1, + }); + + if (blocks.length > 0) { + this.anchorRoot = blocks[0].message.parentRoot; + this.anchorSlot = blocks[0].message.slot; + this.logger.info("Skipped filled epochs", { + from: startEpoch, + to: epoch + 1, + resumeSlot: this.anchorSlot, + }); + } + } + } + + private addPeer = (data: NetworkEventData[NetworkEvent.peerConnected]): void => { + this.logger.debug("BackfillSync: peer connected", {peer: data.peer}); + this.peers.set(data.peer, {failedRequests: 0}); + this.processor.trigger(); + }; + + private removePeer = (data: NetworkEventData[NetworkEvent.peerDisconnected]): void => { + this.peers.delete(data.peer); + if (this.lastGoodPeer === data.peer) { + this.lastGoodPeer = null; + } + }; + + /** + * Pick up to `n` distinct eligible peers, listing the peer that last served us first so a + * hedged fetch reuses its already-warm connection before dialing fresh random peers. + */ + private pickPeers(n: number): PeerIdStr[] { + const eligible = Array.from(this.peers.entries()) + .filter(([_, meta]) => meta.failedRequests < MAX_PEER_FAILURES) + .map(([id]) => id); + + const picked: PeerIdStr[] = []; + if (this.lastGoodPeer !== null && eligible.includes(this.lastGoodPeer)) { + picked.push(this.lastGoodPeer); + } + for (const peer of shuffle(eligible)) { + if (picked.length >= n) break; + if (peer !== this.lastGoodPeer) picked.push(peer); + } + return picked; + } + + private onPeerRequestSuccess(peer: PeerIdStr): void { + const meta = this.peers.get(peer); + if (meta && meta.failedRequests > 0) { + meta.failedRequests = Math.max(0, meta.failedRequests - 1); + } + } + + private onPeerRequestFailed(peer: PeerIdStr): void { + const meta = this.peers.get(peer); + if (meta) { + meta.failedRequests++; + if (meta.failedRequests >= MAX_PEER_FAILURES) { + this.logger.debug("BackfillSync: peer exceeded local failure threshold, excluding from this sync", { + peer, + failRequests: meta.failedRequests, + }); + } + } + } + + close(): void { + this.network.events.off(NetworkEvent.peerConnected, this.addPeer); + this.network.events.off(NetworkEvent.peerDisconnected, this.removePeer); + this.processor.end(new ErrorAborted("BackfillSync")); + this.epochBuffer = []; + this.knownRoots = []; + this.peers.clear(); + } +} diff --git a/packages/beacon-node/src/sync/backfill/index.ts b/packages/beacon-node/src/sync/backfill/index.ts index 7ed7b04bb5f3..3512478f5709 100644 --- a/packages/beacon-node/src/sync/backfill/index.ts +++ b/packages/beacon-node/src/sync/backfill/index.ts @@ -1 +1,2 @@ export * from "./backfill.js"; +export {BackfillSync as BackfillSyncV2} from "./backfillV2.js"; diff --git a/packages/beacon-node/src/sync/options.ts b/packages/beacon-node/src/sync/options.ts index a7248f42e338..0f675f4972ff 100644 --- a/packages/beacon-node/src/sync/options.ts +++ b/packages/beacon-node/src/sync/options.ts @@ -22,6 +22,12 @@ export type SyncOptions = { * allocation to backfill sync. The default of 0 would mean backfill sync will be skipped */ backfillBatchSize: number; + /** + * If false (default), backfill stops at the weak subjectivity period boundary — enough + * for weak subjectivity and the minimum required by the spec. Set true on archival + * nodes that need every block back to genesis. + */ + backfillToGenesis: boolean; /** For testing only, MAX_PENDING_BLOCKS by default */ maxPendingBlocks?: number; @@ -41,5 +47,7 @@ export const defaultSyncOptions: SyncOptions = { disableProcessAsChainSegment: false, /** By default skip the backfill sync */ backfillBatchSize: 0, + /** By default backfill only to the WS period boundary, not all the way to genesis. */ + backfillToGenesis: false, slotImportTolerance: SLOTS_PER_EPOCH, }; diff --git a/packages/beacon-node/test/mocks/mockedBeaconDb.ts b/packages/beacon-node/test/mocks/mockedBeaconDb.ts index 32c453da1800..ec417f2ad0d9 100644 --- a/packages/beacon-node/test/mocks/mockedBeaconDb.ts +++ b/packages/beacon-node/test/mocks/mockedBeaconDb.ts @@ -4,6 +4,7 @@ import {BeaconDb} from "../../src/db/index.js"; import { AttesterSlashingRepository, BLSToExecutionChangeRepository, + BackfillStateRepository, BlobSidecarsArchiveRepository, BlobSidecarsRepository, BlockArchiveRepository, @@ -14,6 +15,7 @@ import { StateArchiveRepository, VoluntaryExitRepository, } from "../../src/db/repositories/index.js"; +import {BackfilledRange} from "../../src/db/single/backfilledRange.js"; export type MockedBeaconDb = Mocked & { block: Mocked; @@ -31,6 +33,9 @@ export type MockedBeaconDb = Mocked & { blsToExecutionChange: Mocked; proposerSlashing: Mocked; attesterSlashing: Mocked; + + backfillState: Mocked; + backfilledRange: Mocked; }; vi.mock("../../src/db/repositories/index.js"); @@ -54,6 +59,9 @@ vi.mock("../../src/db/index.js", async (importActual) => { dataColumnSidecar: vi.mocked(new DataColumnSidecarRepository({} as any, {} as any)), dataColumnSidecarArchive: vi.mocked(new DataColumnSidecarArchiveRepository({} as any, {} as any)), + + backfillState: vi.mocked(new BackfillStateRepository({} as any, {} as any)), + backfilledRange: vi.mocked(new BackfilledRange({} as any, {} as any)), }; }); diff --git a/packages/beacon-node/test/unit/sync/backfill/backfillV2.test.ts b/packages/beacon-node/test/unit/sync/backfill/backfillV2.test.ts new file mode 100644 index 000000000000..623af5271955 --- /dev/null +++ b/packages/beacon-node/test/unit/sync/backfill/backfillV2.test.ts @@ -0,0 +1,371 @@ +import fs from "node:fs"; +import path from "node:path"; +import {fileURLToPath} from "node:url"; +import {afterEach, beforeEach, describe, expect, it, vi} from "vitest"; +import {createBeaconConfig} from "@lodestar/config"; +import {config} from "@lodestar/config/default"; +import {testLogger} from "@lodestar/logger/test-utils"; +import {computeEpochAtSlot} from "@lodestar/state-transition"; +import {phase0, ssz} from "@lodestar/types"; +import {toRootHex} from "@lodestar/utils"; +import {BackfilledRangeWrapper} from "../../../../src/db/single/backfilledRange.js"; +import {INetwork, NetworkEvent, NetworkEventBus} from "../../../../src/network/index.js"; +import {BackfillSync, BackfillSyncEvent} from "../../../../src/sync/backfill/backfillV2.js"; +import {MockedBeaconChain, getMockedBeaconChain} from "../../../mocks/mockedBeaconChain.js"; +import {MockedBeaconDb, getMockedBeaconDb} from "../../../mocks/mockedBeaconDb.js"; + +const __dirname = path.dirname(fileURLToPath(import.meta.url)); + +describe("sync / backfill / backfillV2", () => { + const beaconConfig = createBeaconConfig(config, ssz.Root.defaultValue()); + const logger = testLogger(); + + let chain: MockedBeaconChain; + let db: MockedBeaconDb; + let controller: AbortController; + + beforeEach(() => { + chain = getMockedBeaconChain({config: beaconConfig}); + db = getMockedBeaconDb(); + controller = new AbortController(); + + // BackfilledRange is not auto-mocked by `mockedBeaconDb`; + let currentRange: BackfilledRangeWrapper | null = null; + db.backfilledRange.get = vi.fn(async () => currentRange); + db.backfilledRange.put = vi.fn(async (val) => { + currentRange = val; + }); + db.backfilledRange.delete = vi.fn(async () => { + currentRange = null; + }); + + db.backfillState.get.mockResolvedValue(null); + db.backfillState.put.mockResolvedValue(undefined); + db.blockArchive.batchPut.mockResolvedValue(undefined); + db.blockArchive.values.mockResolvedValue([]); + // No cached blocks — every fetch goes through the network path in tests. + db.blockArchive.getByRoot.mockResolvedValue(null); + db.blockArchive.getSlotByRoot.mockResolvedValue(null); + }); + + afterEach(() => { + controller.abort(); + vi.clearAllMocks(); + }); + + it("should walk the chain by-root using mainnet fixture blocks", {timeout: 10_000}, async () => { + const chainBlocks = getBlocks(); + const {networkEvents, fetchedRoots, walkedPastFixture} = await initSyncFromChain(chainBlocks, { + onMissingRoot: "abort", + }); + + if (!walkedPastFixture) throw new Error("walkedPastFixture not set"); + + connectPeer(networkEvents); + + await withTimeout(walkedPastFixture, 5_000, "backfill chain walk"); + + const expectedRoots = chainBlocks + .slice() + .reverse() + .map((b) => toRootHex(ssz.phase0.BeaconBlock.hashTreeRoot(b.message))); + expect(fetchedRoots.slice(0, chainBlocks.length)).toEqual(expectedRoots); + expect(fetchedRoots.length).toBeGreaterThanOrEqual(chainBlocks.length + 1); + expect(db.backfilledRange.put).toHaveBeenCalled(); + }); + + it("should flush all blocks and complete at genesis", {timeout: 10_000}, async () => { + const chainBlocks = generateLinearChain(0, 3); + const {backfillSync, networkEvents} = await initSyncFromChain(chainBlocks); + + const completed = new Promise((resolve) => { + backfillSync.on(BackfillSyncEvent.completed, () => resolve()); + }); + + connectPeer(networkEvents); + + await withTimeout(completed, 5_000, "backfill completion at genesis"); + + const flushed = db.blockArchive.batchPut.mock.calls.flatMap((call) => call[0]); + expect(flushed).toHaveLength(3); + expect(flushed.map((p: {key: number}) => p.key).sort((a: number, b: number) => a - b)).toEqual([0, 1, 2]); + expect(db.backfillState.put).toHaveBeenCalledWith(0, expect.objectContaining({hasBlock: true})); + }); + + it("should flush at epoch boundary when walking backward", {timeout: 10_000}, async () => { + // Slots 30,31 (epoch 0) and 32,33 (epoch 1). Walk: 33→32→31→30, then anchorRoot is ZERO_HASH and sync completes. + const chainBlocks = generateLinearChain(30, 4); + const {backfillSync, networkEvents} = await initSyncFromChain(chainBlocks); + + const completed = new Promise((resolve) => { + backfillSync.on(BackfillSyncEvent.completed, () => resolve()); + }); + + connectPeer(networkEvents); + + await withTimeout(completed, 5_000, "backfill epoch boundary flush"); + + // Epoch 1 (slots 32,33) should have been flushed when we crossed into epoch 0 + expect(db.backfillState.put).toHaveBeenCalledWith(1, expect.objectContaining({hasBlock: true})); + + const flushed = db.blockArchive.batchPut.mock.calls.flatMap((call) => call[0]); + const epoch1Flushed = flushed.filter((p: {key: number}) => p.key >= 32); + expect(epoch1Flushed).toHaveLength(2); + }); + + it("should flush correctly with skipped slot at epoch boundary", {timeout: 10_000}, async () => { + // Slots 30,31,33 (slot 32 skipped). Walk: 33→31→30, then anchorRoot is ZERO_HASH and sync completes. + // Without the buffer-epoch fix, epoch 1 (slot 33) and epoch 0 (slot 31) would mix in the buffer. + const blocks = generateLinearChainWithSlots([30, 31, 33]); + const {backfillSync, networkEvents} = await initSyncFromChain(blocks); + + const completed = new Promise((resolve) => { + backfillSync.on(BackfillSyncEvent.completed, () => resolve()); + }); + + connectPeer(networkEvents); + + await withTimeout(completed, 5_000, "backfill skipped-slot boundary"); + + // Epoch 1 flush should contain only slot 33 (not mixed with epoch 0) + expect(db.backfillState.put).toHaveBeenCalledWith(1, expect.objectContaining({hasBlock: true})); + + const firstFlush = db.blockArchive.batchPut.mock.calls[0]?.[0] ?? []; + expect(firstFlush).toHaveLength(1); + expect(firstFlush[0].key).toBe(33); + }); + + it("should batch the block_roots window then walk older blocks serially", {timeout: 10_000}, async () => { + // Blocks at slots 0,1,2 then a large gap up to 9000,9001,9002. With anchorState.slot = 9003 + // the block_roots window floor is 9003 - SLOTS_PER_HISTORICAL_ROOT (8192) = 811, so blocks + // 9000-9002 (plus the window-floor entry, which resolves to block 2) are fetched in a Phase 1 + // batch, while blocks 1 and 0 sit below the window and are walked one-by-one in Phase 2. + const chainBlocks = generateLinearChainWithSlots([0, 1, 2, 9000, 9001, 9002]); + const {backfillSync, networkEvents, sendByRoot} = await initSyncFromChain(chainBlocks, {stateSlot: 9003}); + + const completed = new Promise((resolve) => { + backfillSync.on(BackfillSyncEvent.completed, () => resolve()); + }); + + connectPeer(networkEvents); + await withTimeout(completed, 5_000, "backfill two-phase completion"); + + // Phase 1 issues a multi-root batch first; Phase 2 then issues single-root requests. + const callSizes = sendByRoot.mock.calls.map((c) => Array.from(c[1] as Iterable).length); + expect(callSizes[0]).toBeGreaterThan(1); + expect(callSizes.at(-1)).toBe(1); + + const flushed = db.blockArchive.batchPut.mock.calls.flatMap((call) => call[0]); + expect(flushed.map((p: {key: number}) => p.key).sort((a: number, b: number) => a - b)).toEqual([ + 0, 1, 2, 9000, 9001, 9002, + ]); + }); + + it("should short-circuit via the DB cache when blocks are already stored", {timeout: 5_000}, async () => { + // Every block is pre-cached in blockArchive — Phase 1's pre-filter drops the whole + // known-roots window, and Phase 2 walks via the cache short-circuit all the way to + // genesis without a single network call. + const chainBlocks = generateLinearChain(0, 3); + const cachedRoots = indexBlocksByRoot(chainBlocks); + const {backfillSync, networkEvents, sendByRoot} = await initSyncFromChain(chainBlocks, {cachedRoots}); + + const completed = new Promise((resolve) => { + backfillSync.on(BackfillSyncEvent.completed, () => resolve()); + }); + + connectPeer(networkEvents); + await withTimeout(completed, 4_000, "cache short-circuit completion"); + + expect(sendByRoot.mock.calls.length).toBe(0); + }); + + it("should complete despite peers that cannot serve the requested blocks", {timeout: 10_000}, async () => { + // Three peers are connected but two of them serve nothing. The Phase 2 hedged fetch must + // fan out past the dead peers, and Phase 1 must retry past them, so the whole chain still + // gets backfilled via the single peer that has the blocks. + const chainBlocks = generateLinearChainWithSlots([0, 1, 2, 9000, 9001, 9002]); + const {backfillSync, networkEvents} = await initSyncFromChain(chainBlocks, { + stateSlot: 9003, + badPeers: new Set(["bad-peer-1", "bad-peer-2"]), + }); + + const completed = new Promise((resolve) => { + backfillSync.on(BackfillSyncEvent.completed, () => resolve()); + }); + + connectPeers(networkEvents, ["bad-peer-1", "good-peer", "bad-peer-2"]); + await withTimeout(completed, 8_000, "backfill hedged completion"); + + const flushed = db.blockArchive.batchPut.mock.calls.flatMap((call) => call[0]); + expect(flushed.map((p: {key: number}) => p.key).sort((a: number, b: number) => a - b)).toEqual([ + 0, 1, 2, 9000, 9001, 9002, + ]); + }); + + function generateLinearChain(startSlot: number, count: number): phase0.SignedBeaconBlock[] { + const slots = Array.from({length: count}, (_, i) => startSlot + i); + return generateLinearChainWithSlots(slots); + } + + function generateLinearChainWithSlots(slots: number[]): phase0.SignedBeaconBlock[] { + const blocks: phase0.SignedBeaconBlock[] = []; + let parentRoot: Uint8Array = new Uint8Array(32); + for (const slot of slots) { + const block = ssz.phase0.SignedBeaconBlock.defaultValue(); + block.message.slot = slot; + block.message.parentRoot = parentRoot; + blocks.push(block); + parentRoot = Uint8Array.from(ssz.phase0.BeaconBlock.hashTreeRoot(block.message)); + } + return blocks; + } + + function getBlocks(): phase0.SignedBeaconBlock[] { + const json = JSON.parse(fs.readFileSync(path.join(__dirname, "./blocks.json"), "utf-8")) as unknown[]; + return json.map((b) => ssz.phase0.SignedBeaconBlock.fromJson(b)); + } + + function indexBlocksByRoot(blocks: phase0.SignedBeaconBlock[]): Map { + const map = new Map(); + for (const block of blocks) { + map.set(toRootHex(ssz.phase0.BeaconBlock.hashTreeRoot(block.message)), block); + } + return map; + } + + type InitResult = { + backfillSync: BackfillSync; + networkEvents: NetworkEventBus; + fetchedRoots: string[]; + sendByRoot: ReturnType; + walkedPastFixture?: Promise; + }; + + async function initSyncFromChain( + chainBlocks: phase0.SignedBeaconBlock[], + opts?: { + onMissingRoot?: "abort"; + stateSlot?: number; + badPeers?: Set; + cachedRoots?: Map; + } + ): Promise { + const blocksByRoot = indexBlocksByRoot(chainBlocks); + // biome-ignore lint/style/noNonNullAssertion: chain always has blocks + const tip = chainBlocks.at(-1)!; + const anchorRoot = ssz.phase0.BeaconBlock.hashTreeRoot(tip.message); + const anchorSlot = tip.message.slot; + + // Mirror BeaconState.block_roots: ascending (slot, root) pairs so the anchor-state mock + // can answer getBlockRootAtSlot() with the most recent block at or before a given slot. + const sortedRoots = chainBlocks + .map((b) => ({slot: b.message.slot, root: ssz.phase0.BeaconBlock.hashTreeRoot(b.message)})) + .sort((a, b) => a.slot - b.slot); + const stateSlot = opts?.stateSlot ?? anchorSlot + 1; + function getBlockRootAtSlot(slot: number): Uint8Array { + let result = sortedRoots[0].root; + for (const entry of sortedRoots) { + if (entry.slot > slot) break; + result = entry.root; + } + return result; + } + + let onWalkedPastFixture: (() => void) | undefined; + let walkedPastFixture: Promise | undefined; + if (opts?.onMissingRoot === "abort") { + walkedPastFixture = new Promise((resolve) => { + onWalkedPastFixture = () => { + onWalkedPastFixture = undefined; + resolve(); + controller.abort(); + }; + }); + } + + const fetchedRoots: string[] = []; + const networkEvents = new NetworkEventBus(); + + const sendByRoot = vi.fn(async (peerId: unknown, roots: Iterable) => { + const out: phase0.SignedBeaconBlock[] = []; + // A "bad" peer is reachable but serves no blocks (mirrors a peer that pruned history). + if (opts?.badPeers?.has(String(peerId))) { + return out; + } + for (const root of roots) { + const hex = toRootHex(root); + fetchedRoots.push(hex); + const block = blocksByRoot.get(hex); + if (block) { + out.push(block); + } else { + onWalkedPastFixture?.(); + } + } + return out; + }); + + const network: Partial = { + events: networkEvents, + sendBeaconBlocksByRoot: sendByRoot, + reportPeer: () => {}, + }; + + if (opts?.cachedRoots) { + const cached = opts.cachedRoots; + db.blockArchive.getByRoot = vi.fn(async (root: Uint8Array) => cached.get(toRootHex(root)) ?? null); + db.blockArchive.getSlotByRoot = vi.fn(async (root: Uint8Array) => { + const block = cached.get(toRootHex(root)); + return block ? block.message.slot : null; + }); + } + + const anchorState = { + slot: stateSlot, + latestBlockHeader: {slot: anchorSlot} as phase0.BeaconBlockHeader, + getBlockRootAtSlot, + computeAnchorCheckpoint: () => ({ + checkpoint: {epoch: computeEpochAtSlot(anchorSlot), root: anchorRoot}, + blockHeader: {slot: anchorSlot} as phase0.BeaconBlockHeader, + }), + }; + + const backfillSync = await BackfillSync.init( + {backfillBatchSize: 64, backfillToGenesis: true}, + { + chain, + db, + network: network as INetwork, + config: beaconConfig, + logger, + metrics: null, + anchorState: anchorState as any, + signal: controller.signal, + } + ); + + return {backfillSync, networkEvents, fetchedRoots, sendByRoot, walkedPastFixture}; + } + + function connectPeer(networkEvents: NetworkEventBus): void { + connectPeers(networkEvents, ["test-peer"]); + } + + function connectPeers(networkEvents: NetworkEventBus, peerIds: string[]): void { + for (const peer of peerIds) { + networkEvents.emit(NetworkEvent.peerConnected, { + peer, + status: {} as any, + custodyColumns: [], + clientAgent: "test-client", + }); + } + } + + function withTimeout(promise: Promise, ms: number, label: string): Promise { + return Promise.race([ + promise, + new Promise((_, reject) => setTimeout(() => reject(new Error(`Timed out: ${label}`)), ms)), + ]); + } +}); diff --git a/packages/cli/src/options/beaconNodeOptions/sync.ts b/packages/cli/src/options/beaconNodeOptions/sync.ts index a33fa43db470..87ba46b29a9c 100644 --- a/packages/cli/src/options/beaconNodeOptions/sync.ts +++ b/packages/cli/src/options/beaconNodeOptions/sync.ts @@ -6,6 +6,7 @@ export type SyncArgs = { "sync.disableProcessAsChainSegment"?: boolean; "sync.disableRangeSync"?: boolean; "sync.backfillBatchSize"?: number; + "sync.backfillToGenesis"?: boolean; "sync.slotImportTolerance"?: number; }; @@ -14,6 +15,7 @@ export function parseArgs(args: SyncArgs): IBeaconNodeOptions["sync"] { isSingleNode: args["sync.isSingleNode"], disableProcessAsChainSegment: args["sync.disableProcessAsChainSegment"], backfillBatchSize: args["sync.backfillBatchSize"] ?? defaultOptions.sync.backfillBatchSize, + backfillToGenesis: args["sync.backfillToGenesis"] ?? defaultOptions.sync.backfillToGenesis, disableRangeSync: args["sync.disableRangeSync"], slotImportTolerance: args["sync.slotImportTolerance"] ?? defaultOptions.sync.slotImportTolerance, }; @@ -62,4 +64,12 @@ Use only for local networks with a single node, can be dangerous in regular netw defaultDescription: String(defaultOptions.sync.backfillBatchSize), group: "sync", }, + + "sync.backfillToGenesis": { + type: "boolean", + description: + "Backfill all the way to genesis. Off by default: backfill stops at the weak subjectivity period boundary, which is the minimum required by the spec. Enable for archival nodes.", + defaultDescription: String(defaultOptions.sync.backfillToGenesis), + group: "sync", + }, };