feat: n historical states #6008
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -54,13 +54,11 @@ export class Archiver { | |
|
|
||
| if (!opts.disableArchiveOnCheckpoint) { | ||
| this.chain.emitter.on(ChainEvent.forkChoiceFinalized, this.onFinalizedCheckpoint); | ||
|
|
||
| signal.addEventListener( | ||
| "abort", | ||
| () => { | ||
| this.chain.emitter.off(ChainEvent.forkChoiceFinalized, this.onFinalizedCheckpoint); | ||
| }, | ||
| {once: true} | ||
| ); | ||
|
|
@@ -76,15 +74,6 @@ export class Archiver { | |
| return this.jobQueue.push(finalized); | ||
| }; | ||
|
|
||
| private processFinalizedCheckpoint = async (finalized: CheckpointWithHex): Promise<void> => { | ||
| try { | ||
| const finalizedEpoch = finalized.epoch; | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -15,7 +15,7 @@ import {ZERO_HASH_HEX} from "../../constants/index.js"; | |
| import {toCheckpointHex} from "../stateCache/index.js"; | ||
| import {isOptimisticBlock} from "../../util/forkChoice.js"; | ||
| import {isQueueErrorAborted} from "../../util/queue/index.js"; | ||
| import {ReorgEventData} from "../emitter.js"; | ||
| import {REPROCESS_MIN_TIME_TO_NEXT_SLOT_SEC} from "../reprocess.js"; | ||
| import type {BeaconChain} from "../chain.js"; | ||
| import {FullyVerifiedBlock, ImportBlockOpts, AttestationImportOpt} from "./types.js"; | ||
|
|
@@ -62,6 +62,7 @@ export async function importBlock( | |
| const blockRootHex = toHexString(blockRoot); | ||
| const currentEpoch = computeEpochAtSlot(this.forkChoice.getTime()); | ||
| const blockEpoch = computeEpochAtSlot(block.message.slot); | ||
| const parentEpoch = computeEpochAtSlot(parentBlockSlot); | ||
| const prevFinalizedEpoch = this.forkChoice.getFinalizedCheckpoint().epoch; | ||
| const blockDelaySec = (fullyVerifiedBlock.seenTimestampSec - postState.genesisTime) % this.config.SECONDS_PER_SLOT; | ||
|
|
||
|
|
@@ -202,16 +203,15 @@ export async function importBlock( | |
| } | ||
| } | ||
|
|
||
| // 5. Compute head, always add to state cache so that it'll not be pruned soon | ||
|
|
||
| const oldHead = this.forkChoice.getHead(); | ||
| const newHead = this.recomputeForkChoiceHead(); | ||
| const currFinalizedEpoch = this.forkChoice.getFinalizedCheckpoint().epoch; | ||
|
|
||
| // always set head state so it'll never be pruned from state cache | ||
| this.regen.updateHeadState(newHead.stateRoot, postState); | ||
| if (newHead.blockRoot !== oldHead.blockRoot) { | ||
| this.emitter.emit(routes.events.EventType.head, { | ||
| block: newHead.blockRoot, | ||
| epochTransition: computeStartSlotAtEpoch(computeEpochAtSlot(newHead.slot)) === newHead.slot, | ||
|
|
@@ -331,12 +331,17 @@ export async function importBlock( | |
| this.logger.verbose("After importBlock caching postState without SSZ cache", {slot: postState.slot}); | ||
| } | ||
|
|
||
| if (parentEpoch < blockEpoch) { | ||
| this.shufflingCache.processState(postState); | ||
| this.logger.verbose("Processed shuffling for next epoch", {parentEpoch, blockEpoch, slot: block.message.slot}); | ||
|
|
||
| // This is the real check point state per spec because the root is in current epoch | ||
| // it's important to add this to cache, when chain is finalized we'll query this state later | ||
| const checkpointState = postState; | ||
| const cp = getCheckpointFromState(checkpointState); | ||
|
There was a problem hiding this comment. put this inside |
||
| if (block.message.slot % SLOTS_PER_EPOCH === 0) { | ||
twoeths marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| this.regen.addCheckpointState(cp, checkpointState); | ||
| } | ||
|
|
||
| // Note: in-lined code from previos handler of ChainEvent.checkpoint | ||
| this.logger.verbose("Checkpoint processed", toCheckpointHex(cp)); | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -39,7 +39,7 @@ import {IExecutionEngine, IExecutionBuilder} from "../execution/index.js"; | |
| import {Clock, ClockEvent, IClock} from "../util/clock.js"; | ||
| import {ensureDir, writeIfNotExist} from "../util/file.js"; | ||
| import {isOptimisticBlock} from "../util/forkChoice.js"; | ||
| import {CHECKPOINT_STATES_FOLDER, PersistentCheckpointStateCache, StateContextCache} from "./stateCache/index.js"; | ||
| import {BlockProcessor, ImportBlockOpts} from "./blocks/index.js"; | ||
| import {ChainEventEmitter, ChainEvent} from "./emitter.js"; | ||
| import {IBeaconChain, ProposerPreparationData, BlockHash, StateGetOpts} from "./interface.js"; | ||
|
|
@@ -75,6 +75,10 @@ import {BlockAttributes, produceBlockBody} from "./produceBlock/produceBlockBody | |
| import {computeNewStateRoot} from "./produceBlock/computeNewStateRoot.js"; | ||
| import {BlockInput} from "./blocks/types.js"; | ||
| import {SeenAttestationDatas} from "./seenCache/seenAttestationData.js"; | ||
| import {ShufflingCache} from "./shufflingCache.js"; | ||
| import {MemoryCheckpointStateCache} from "./stateCache/memoryCheckpointsCache.js"; | ||
| import {FilePersistentApis} from "./stateCache/persistent/file.js"; | ||
| import {DbPersistentApis} from "./stateCache/persistent/db.js"; | ||
|
|
||
| /** | ||
| * Arbitrary constants, blobs should be consumed immediately in the same slot they are produced. | ||
|
|
@@ -130,6 +134,7 @@ export class BeaconChain implements IBeaconChain { | |
|
|
||
| readonly beaconProposerCache: BeaconProposerCache; | ||
| readonly checkpointBalancesCache: CheckpointBalancesCache; | ||
| readonly shufflingCache: ShufflingCache; | ||
| // TODO DENEB: Prune data structure every time period, for both old entries | ||
| /** Map keyed by executionPayload.blockHash of the block for those blobs */ | ||
| readonly producedBlobSidecarsCache = new Map<BlockHash, {blobSidecars: deneb.BlobSidecars; slot: Slot}>(); | ||
|
|
@@ -211,6 +216,7 @@ export class BeaconChain implements IBeaconChain { | |
|
|
||
| this.beaconProposerCache = new BeaconProposerCache(opts); | ||
| this.checkpointBalancesCache = new CheckpointBalancesCache(); | ||
| this.shufflingCache = new ShufflingCache(); | ||
|
|
||
| // Restore state caches | ||
| // anchorState may already by a CachedBeaconState. If so, don't create the cache again, since deserializing all | ||
|
|
@@ -230,12 +236,26 @@ export class BeaconChain implements IBeaconChain { | |
| this.pubkey2index = cachedState.epochCtx.pubkey2index; | ||
| this.index2pubkey = cachedState.epochCtx.index2pubkey; | ||
|
|
||
| const stateCache = new StateContextCache(this.opts, {metrics}); | ||
| const persistentApis = this.opts.persistCheckpointStatesToFile | ||
| ? new FilePersistentApis(CHECKPOINT_STATES_FOLDER) | ||
| : new DbPersistentApis(this.db); | ||
| const checkpointStateCache = this.opts.persistentCheckpointStateCache | ||
| ? new PersistentCheckpointStateCache( | ||
| { | ||
| metrics, | ||
| logger, | ||
| clock, | ||
| shufflingCache: this.shufflingCache, | ||
| getHeadState: this.getHeadState.bind(this), | ||
| persistentApis, | ||
| }, | ||
| this.opts | ||
| ) | ||
| : new MemoryCheckpointStateCache({metrics}); | ||
|
|
||
| const {checkpoint} = computeAnchorCheckpoint(config, anchorState); | ||
| stateCache.add(cachedState); | ||
| checkpointStateCache.add(checkpoint, cachedState); | ||
|
|
||
| const forkChoice = initializeForkChoice( | ||
|
|
@@ -841,15 +861,19 @@ export class BeaconChain implements IBeaconChain { | |
| this.logger.verbose("Fork choice justified", {epoch: cp.epoch, root: cp.rootHex}); | ||
| } | ||
|
|
||
| private async onForkChoiceFinalized(this: BeaconChain, cp: CheckpointWithHex): Promise<void> { | ||
| this.logger.verbose("Fork choice finalized", {epoch: cp.epoch, root: cp.rootHex}); | ||
| this.seenBlockProposers.prune(computeStartSlotAtEpoch(cp.epoch)); | ||
|
|
||
| // TODO: Improve using regen here | ||
| const headState = this.regen.getStateSync(this.forkChoice.getHead().stateRoot); | ||
| // the finalized state could be from disk | ||
| const finalizedStateOrBytes = await this.regen.getCheckpointStateOrBytes(cp); | ||
| if (!finalizedStateOrBytes) { | ||
| throw Error("No state in cache for finalized checkpoint state epoch #" + cp.epoch); | ||
| } | ||
| if (headState) { | ||
| this.opPool.pruneAll(headState, finalizedStateOrBytes); | ||
| } | ||
| } | ||
|
|
||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -14,8 +14,13 @@ import { | |
| BLS_WITHDRAWAL_PREFIX, | ||
| } from "@lodestar/params"; | ||
| import {Epoch, phase0, capella, ssz, ValidatorIndex} from "@lodestar/types"; | ||
| import {ChainForkConfig} from "@lodestar/config"; | ||
| import {IBeaconDb} from "../../db/index.js"; | ||
| import {SignedBLSToExecutionChangeVersioned} from "../../util/types.js"; | ||
| import { | ||
| getValidatorsBytesFromStateBytes, | ||
| getWithdrawalCredentialFirstByteFromValidatorBytes, | ||
| } from "../../util/sszBytes.js"; | ||
| import {isValidBlsToExecutionChangeForBlockInclusion} from "./utils.js"; | ||
|
|
||
| type HexRoot = string; | ||
|
|
@@ -270,11 +275,11 @@ export class OpPool { | |
| /** | ||
| * Prune all types of transactions given the latest head state | ||
| */ | ||
| pruneAll(headState: CachedBeaconStateAllForks, finalizedState: CachedBeaconStateAllForks | Uint8Array): void { | ||
| this.pruneAttesterSlashings(headState); | ||
| this.pruneProposerSlashings(headState); | ||
| this.pruneVoluntaryExits(headState); | ||
| this.pruneBlsToExecutionChanges(headState.config, finalizedState); | ||
| } | ||
|
|
||
| /** | ||
|
|
@@ -344,17 +349,34 @@ export class OpPool { | |
| * credentials | ||
| */ | ||
| private pruneBlsToExecutionChanges( | ||
| config: ChainForkConfig, | ||
| finalizedStateOrBytes: CachedBeaconStateAllForks | Uint8Array | ||
| ): void { | ||
| const validatorBytes = | ||
| finalizedStateOrBytes instanceof Uint8Array | ||
| ? getValidatorsBytesFromStateBytes(config, finalizedStateOrBytes) | ||
| : null; | ||
|
|
||
| for (const [key, blsToExecutionChange] of this.blsToExecutionChanges.entries()) { | ||
| // there are at least finalied state bytes | ||
| let withDrawableCredentialFirstByte: number | null; | ||
twoeths marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| const validatorIndex = blsToExecutionChange.data.message.validatorIndex; | ||
| if (finalizedStateOrBytes instanceof Uint8Array) { | ||
| if (!validatorBytes) { | ||
| throw Error( | ||
| "Not able to extract validator bytes from finalized state bytes with length " + finalizedStateOrBytes.length | ||
| ); | ||
| } | ||
| withDrawableCredentialFirstByte = getWithdrawalCredentialFirstByteFromValidatorBytes( | ||
| validatorBytes, | ||
| validatorIndex | ||
| ); | ||
| } else { | ||
| const validator = finalizedStateOrBytes.validators.getReadonly(validatorIndex); | ||
| withDrawableCredentialFirstByte = validator.withdrawalCredentials[0]; | ||
| } | ||
| if (withDrawableCredentialFirstByte !== BLS_WITHDRAWAL_PREFIX) { | ||
| this.blsToExecutionChanges.delete(key); | ||
| } | ||
| } | ||
| } | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -3,12 +3,16 @@ import {defaultOptions as defaultValidatorOptions} from "@lodestar/validator"; | |
| import {ArchiverOpts} from "./archiver/index.js"; | ||
| import {ForkChoiceOpts} from "./forkChoice/index.js"; | ||
| import {LightClientServerOpts} from "./lightClient/index.js"; | ||
| import {PersistentCheckpointStateCacheOpts} from "./stateCache/types.js"; | ||
| import {StateContextCacheOpts} from "./stateCache/stateContextCache.js"; | ||
|
|
||
| export type IChainOptions = BlockProcessOpts & | ||
| PoolOpts & | ||
| SeenCacheOpts & | ||
| ForkChoiceOpts & | ||
| ArchiverOpts & | ||
| StateContextCacheOpts & | ||
| PersistentCheckpointStateCacheOpts & | ||
| LightClientServerOpts & { | ||
| blsVerifyAllMainThread?: boolean; | ||
| blsVerifyAllMultiThread?: boolean; | ||
|
|
@@ -27,6 +31,10 @@ export type IChainOptions = BlockProcessOpts & | |
| trustedSetup?: string; | ||
| broadcastValidationStrictness?: string; | ||
| minSameMessageSignatureSetsToBatch: number; | ||
| // TODO: change to n_historical_states | ||
| persistentCheckpointStateCache?: boolean; | ||
| /** by default persist checkpoint state to db */ | ||
| persistCheckpointStatesToFile?: boolean; | ||
| }; | ||
|
|
||
| export type BlockProcessOpts = { | ||
|
|
@@ -88,4 +96,14 @@ export const defaultChainOptions: IChainOptions = { | |
| // batching too much may block the I/O thread so if useWorker=false, suggest this value to be 32 | ||
| // since this batch attestation work is designed to work with useWorker=true, make this the lowest value | ||
| minSameMessageSignatureSetsToBatch: 2, | ||
| // TODO: change to false, leaving here to ease testing | ||
| persistentCheckpointStateCache: true, | ||
| // TODO: change to false, leaving here to ease testing | ||
| persistCheckpointStatesToFile: true, | ||
|
|
||
| // since Sep 2023, only cache up to 32 states by default. If a big reorg happens it'll load checkpoint state from disk and regen from there. | ||
| // TODO: change to 128, leaving here to ease testing | ||
| maxStates: 32, | ||
| // only used when persistentCheckpointStateCache = true | ||
| maxEpochsInMemory: 2, | ||
| }; | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -83,7 +83,8 @@ export class PrepareNextSlotScheduler { | |
| headSlot, | ||
| clockSlot, | ||
| }); | ||
| // It's important to still do this to get through Holesky unfinality time of low resouce nodes | ||
| await this.prunePerSlot(clockSlot); | ||
| return; | ||
| } | ||
|
|
||
|
|
@@ -173,11 +174,29 @@ export class PrepareNextSlotScheduler { | |
| this.chain.emitter.emit(routes.events.EventType.payloadAttributes, {data, version: fork}); | ||
| } | ||
| } | ||
|
|
||
| // do this after all as it's the lowest priority task | ||
| await this.prunePerSlot(clockSlot); | ||
| } catch (e) { | ||
| if (!isErrorAborted(e) && !isQueueErrorAborted(e)) { | ||
| this.metrics?.precomputeNextEpochTransition.count.inc({result: "error"}, 1); | ||
| this.logger.error("Failed to run prepareForNextSlot", {nextEpoch, isEpochTransition, prepareSlot}, e as Error); | ||
| } | ||
| } | ||
| }; | ||
|
|
||
| /** | ||
| * Pruning at the last 1/3 slot of epoch is the safest time because all epoch transitions already use the checkpoint states cached | ||
| * one down side of this is when `inMemoryEpochs = 0` and gossip block hasn't come yet then we have to reload state we added 2/3 slot ago | ||
| * however, it's not likely `inMemoryEpochs` is configured as 0, and this scenario rarely happen | ||
| * since we only use `inMemoryEpochs = 0` for testing, if it happens it's a good thing because it helps us test the reload flow | ||
| */ | ||
| private prunePerSlot = async (clockSlot: Slot): Promise<void> => { | ||
| // a contabo vpss can have 10-12 holesky epoch transitions per epoch when syncing, stronger node may have more | ||
| // although it can survive during syncing if we prune per epoch, it's better to prune at the last 1/3 of every slot | ||
| // at synced time, it's likely we only prune at the 1st slot of epoch, all other prunes are no-op | ||
| const nextEpoch = computeEpochAtSlot(clockSlot) + 1; | ||
| const pruneCount = await this.chain.regen.pruneCheckpointStateCache(); | ||
| this.logger.verbose("Pruned checkpoint state cache", {clockSlot, nextEpoch, pruneCount}); | ||
| }; | ||
| } | ||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Can you add a jsdoc comment?