diff --git a/packages/api/test/unit/builder/testData.ts b/packages/api/test/unit/builder/testData.ts index aa9082c16589..149e164187f5 100644 --- a/packages/api/test/unit/builder/testData.ts +++ b/packages/api/test/unit/builder/testData.ts @@ -23,7 +23,7 @@ export const testData: GenericServerTestCases = { }, submitBlindedBlock: { args: {signedBlindedBlock: {data: ssz.electra.SignedBlindedBeaconBlock.defaultValue()}}, - res: {data: ssz.deneb.ExecutionPayloadAndBlobsBundle.defaultValue(), meta: {version: ForkName.electra}}, + res: {data: ssz.electra.ExecutionPayloadAndBlobsBundle.defaultValue(), meta: {version: ForkName.electra}}, }, submitBlindedBlockV2: { args: {signedBlindedBlock: {data: ssz.fulu.SignedBlindedBeaconBlock.defaultValue()}}, diff --git a/packages/beacon-node/src/api/impl/validator/index.ts b/packages/beacon-node/src/api/impl/validator/index.ts index d5b36043db33..ecc3eac51b1c 100644 --- a/packages/beacon-node/src/api/impl/validator/index.ts +++ b/packages/beacon-node/src/api/impl/validator/index.ts @@ -14,6 +14,7 @@ import { isForkPostBellatrix, isForkPostDeneb, isForkPostElectra, + isForkPostFulu, isForkPostGloas, } from "@lodestar/params"; import { @@ -1124,26 +1125,33 @@ export function getValidatorApi( async getProposerDuties({epoch}, _context, opts?: {v2?: boolean}) { notWhileSyncing(); - // Early check that epoch is no more than current_epoch + 1, or allow for pre-genesis const currentEpoch = currentEpochWithDisparity(); const nextEpoch = currentEpoch + 1; - if (currentEpoch >= 0 && epoch > nextEpoch) { + const startSlot = computeStartSlotAtEpoch(epoch); + const prepareNextSlotLookAheadMs = + config.SLOT_DURATION_MS - config.getSlotComponentDurationMs(PREPARE_NEXT_SLOT_BPS); + const toNextEpochMs = msToNextEpoch(); + const nearNextEpoch = toNextEpochMs < prepareNextSlotLookAheadMs; + // Post-Fulu the proposer lookahead is deterministic and known a full epoch ahead, so + // close to the boundary `currentEpoch + 2` is serveable from the upcoming-epoch + // checkpoint state (its `nextProposers`). Pre-Fulu / mid-epoch: `currentEpoch + 1` max. + const isPostFulu = isForkPostFulu(config.getForkName(startSlot)); + const maxFutureEpoch = isPostFulu && nearNextEpoch && opts?.v2 ? nextEpoch + 1 : nextEpoch; + if (currentEpoch >= 0 && epoch > maxFutureEpoch) { throw new ApiError(400, `Requested epoch ${epoch} must not be more than one epoch in the future`); } const head = chain.forkChoice.getHead(); let state: IBeaconStateView | undefined = undefined; - const startSlot = computeStartSlotAtEpoch(epoch); - const prepareNextSlotLookAheadMs = - config.SLOT_DURATION_MS - config.getSlotComponentDurationMs(PREPARE_NEXT_SLOT_BPS); - const toNextEpochMs = msToNextEpoch(); // validators may request next epoch's duties when it's close to next epoch - // this is to avoid missed block proposal due to 0 epoch look ahead - if (epoch === nextEpoch && toNextEpochMs < prepareNextSlotLookAheadMs) { + // this is to avoid missed block proposal due to 0 epoch look ahead. + // Post-Fulu, `nextEpoch + 1` is served from the same upcoming-epoch (`nextEpoch`) + // checkpoint state via its `nextProposers` (deterministic proposer lookahead). + if (nearNextEpoch && (epoch === nextEpoch || (isPostFulu && epoch === nextEpoch + 1))) { // wait for maximum 1 slot for cp state which is the timeout of validator api const cpState = await waitForCheckpointState({ rootHex: head.blockRoot, - epoch, + epoch: nextEpoch, }); if (cpState) { state = cpState; @@ -1218,7 +1226,7 @@ export function getValidatorApi( // It should be set to the latest block applied to `self` or the genesis block root. const dependentRoot = // In v2 the dependent root is different after fulu due to deterministic proposer lookahead - proposerShufflingDecisionRoot(opts?.v2 ? config.getForkName(startSlot) : ForkName.phase0, state) || + proposerShufflingDecisionRoot(opts?.v2 ? config.getForkName(startSlot) : ForkName.phase0, state, epoch) || (await getGenesisBlockRoot(state)); return { diff --git a/packages/beacon-node/test/unit/api/impl/validator/duties/proposer.test.ts b/packages/beacon-node/test/unit/api/impl/validator/duties/proposer.test.ts index e31a6f7fd772..b389bbb6ca15 100644 --- a/packages/beacon-node/test/unit/api/impl/validator/duties/proposer.test.ts +++ b/packages/beacon-node/test/unit/api/impl/validator/duties/proposer.test.ts @@ -1,7 +1,7 @@ import {afterEach, beforeEach, describe, expect, it, vi} from "vitest"; import {routes} from "@lodestar/api"; -import {config} from "@lodestar/config/default"; -import {MAX_EFFECTIVE_BALANCE, SLOTS_PER_EPOCH} from "@lodestar/params"; +import {getConfig} from "@lodestar/config/test-utils"; +import {ForkName, MAX_EFFECTIVE_BALANCE, SLOTS_PER_EPOCH} from "@lodestar/params"; import {BeaconStateAllForks, BeaconStateView} from "@lodestar/state-transition"; import {Slot} from "@lodestar/types"; import {SYNC_TOLERANCE_EPOCHS, getValidatorApi} from "../../../../../../src/api/impl/validator/index.js"; @@ -14,6 +14,8 @@ import {generateState, zeroProtoBlock} from "../../../../../utils/state.js"; import {generateValidators} from "../../../../../utils/validator.js"; describe("get proposers api impl", () => { + // mainnet is post-Fulu; test the realistic deterministic-proposer-lookahead path + const config = getConfig(ForkName.fulu); const currentEpoch = 2; const currentSlot = SLOTS_PER_EPOCH * currentEpoch; @@ -25,7 +27,7 @@ describe("get proposers api impl", () => { beforeEach(() => { vi.useFakeTimers({now: 0}); vi.advanceTimersByTime(currentSlot * config.SLOT_DURATION_MS); - modules = getApiTestModules({clock: "real"}); + modules = getApiTestModules({clock: "real", config}); api = getValidatorApi(defaultApiOptions, modules); initializeState(currentSlot); @@ -66,18 +68,18 @@ describe("get proposers api impl", () => { vi.advanceTimersByTime((SYNC_TOLERANCE_EPOCHS * SLOTS_PER_EPOCH + 1) * config.SLOT_DURATION_MS); vi.spyOn(modules.sync, "state", "get").mockReturnValue(SyncState.SyncingHead); - await expect(api.getProposerDuties({epoch: 1})).rejects.toThrow("Node is syncing - headSlot 0 currentSlot 97"); + await expect(api.getProposerDutiesV2({epoch: 1})).rejects.toThrow("Node is syncing - headSlot 0 currentSlot 97"); }); it("should raise error if node stalled", async () => { vi.advanceTimersByTime((SYNC_TOLERANCE_EPOCHS * SLOTS_PER_EPOCH + 1) * config.SLOT_DURATION_MS); vi.spyOn(modules.sync, "state", "get").mockReturnValue(SyncState.Stalled); - await expect(api.getProposerDuties({epoch: 1})).rejects.toThrow("Node is syncing - waiting for peers"); + await expect(api.getProposerDutiesV2({epoch: 1})).rejects.toThrow("Node is syncing - waiting for peers"); }); it("should get proposers for current epoch", async () => { - const {data: result} = (await api.getProposerDuties({epoch: currentEpoch})) as { + const {data: result} = (await api.getProposerDutiesV2({epoch: currentEpoch})) as { data: routes.validator.ProposerDutyList; }; @@ -92,7 +94,7 @@ describe("get proposers api impl", () => { it("should get proposers for next epoch", async () => { const nextEpoch = currentEpoch + 1; - const {data: result} = (await api.getProposerDuties({epoch: nextEpoch})) as { + const {data: result} = (await api.getProposerDutiesV2({epoch: nextEpoch})) as { data: routes.validator.ProposerDutyList; }; @@ -114,7 +116,7 @@ describe("get proposers api impl", () => { finalized: true, }); - const {data: result} = (await api.getProposerDuties({epoch: historicalEpoch})) as { + const {data: result} = (await api.getProposerDutiesV2({epoch: historicalEpoch})) as { data: routes.validator.ProposerDutyList; }; @@ -126,16 +128,16 @@ describe("get proposers api impl", () => { }); it("should raise error for more than one epoch in the future", async () => { - await expect(api.getProposerDuties({epoch: currentEpoch + 2})).rejects.toThrow( + await expect(api.getProposerDutiesV2({epoch: currentEpoch + 2})).rejects.toThrow( "Requested epoch 4 must not be more than one epoch in the future" ); }); it("should have different proposer validator public keys for current and next epoch", async () => { - const {data: currentProposers} = (await api.getProposerDuties({epoch: currentEpoch})) as { + const {data: currentProposers} = (await api.getProposerDutiesV2({epoch: currentEpoch})) as { data: routes.validator.ProposerDutyList; }; - const {data: nextProposers} = (await api.getProposerDuties({epoch: currentEpoch + 1})) as { + const {data: nextProposers} = (await api.getProposerDutiesV2({epoch: currentEpoch + 1})) as { data: routes.validator.ProposerDutyList; }; @@ -144,10 +146,10 @@ describe("get proposers api impl", () => { }); it("should have different proposer validator indexes for current and next epoch", async () => { - const {data: currentProposers} = (await api.getProposerDuties({epoch: currentEpoch})) as { + const {data: currentProposers} = (await api.getProposerDutiesV2({epoch: currentEpoch})) as { data: routes.validator.ProposerDutyList; }; - const {data: nextProposers} = (await api.getProposerDuties({epoch: currentEpoch + 1})) as { + const {data: nextProposers} = (await api.getProposerDutiesV2({epoch: currentEpoch + 1})) as { data: routes.validator.ProposerDutyList; }; @@ -155,10 +157,10 @@ describe("get proposers api impl", () => { }); it("should have different proposer slots for current and next epoch", async () => { - const {data: currentProposers} = (await api.getProposerDuties({epoch: currentEpoch})) as { + const {data: currentProposers} = (await api.getProposerDutiesV2({epoch: currentEpoch})) as { data: routes.validator.ProposerDutyList; }; - const {data: nextProposers} = (await api.getProposerDuties({epoch: currentEpoch + 1})) as { + const {data: nextProposers} = (await api.getProposerDutiesV2({epoch: currentEpoch + 1})) as { data: routes.validator.ProposerDutyList; }; diff --git a/packages/beacon-node/test/utils/api.ts b/packages/beacon-node/test/utils/api.ts index 975e2bc7bacf..3950cbf606fe 100644 --- a/packages/beacon-node/test/utils/api.ts +++ b/packages/beacon-node/test/utils/api.ts @@ -23,7 +23,7 @@ export function getApiTestModules(opts?: Partial): Api const networkStub = getMockedNetwork(); return { - config, + config: opts?.config ?? config, chain: chainStub, sync: syncStub, db: dbStub, diff --git a/packages/state-transition/src/util/shuffling.ts b/packages/state-transition/src/util/shuffling.ts index 56ebfb79b8fe..9710b2b3bf07 100644 --- a/packages/state-transition/src/util/shuffling.ts +++ b/packages/state-transition/src/util/shuffling.ts @@ -17,29 +17,31 @@ import {computeStartSlotAtEpoch} from "./epoch.js"; import {EpochShuffling} from "./epochShuffling.js"; /** - * Returns the block root which decided the proposer shuffling for the current epoch. This root - * can be used to key this proposer shuffling. + * Block root that decided the proposer shuffling for `proposalEpoch` (keys that shuffling). + * Computed for the requested epoch, not `state.epoch`, so it is correct when `state` is one + * epoch off the requested epoch (e.g. serving next-epoch duties from the current state). * - * Returns `null` on the one-off scenario where the genesis block decides its own shuffling. - * It should be set to the latest block applied to this `state` or the genesis block root. + * Returns `null` when the genesis block decides its own shuffling (caller falls back to the + * genesis block root). */ -export function proposerShufflingDecisionRoot(fork: ForkName, state: IBeaconStateView): Root | null { - const decisionSlot = proposerShufflingDecisionSlot(fork, state); +export function proposerShufflingDecisionRoot( + fork: ForkName, + state: IBeaconStateView, + proposalEpoch: Epoch +): Root | null { + const decisionSlot = proposerShufflingDecisionSlot(fork, proposalEpoch); if (state.slot === decisionSlot) { return null; } return state.getBlockRootAtSlot(decisionSlot); } -/** - * Returns the slot at which the proposer shuffling was decided. The block root at this slot - * can be used to key the proposer shuffling for the current epoch. - */ -function proposerShufflingDecisionSlot(fork: ForkName, state: IBeaconStateView): Slot { - // After fulu, the decision slot is in previous epoch due to deterministic proposer lookahead - const epoch = isForkPostFulu(fork) ? state.epoch - 1 : state.epoch; - const startSlot = computeStartSlotAtEpoch(epoch); - return Math.max(startSlot - 1, 0); +/** Slot whose block root keys the proposer shuffling for `proposalEpoch`. */ +function proposerShufflingDecisionSlot(fork: ForkName, proposalEpoch: Epoch): Slot { + // Post-Fulu the shuffling is decided one epoch earlier (deterministic proposer lookahead, + // MIN_SEED_LOOKAHEAD = 1); pre-Fulu it is the last block before `proposalEpoch`. + const decisionEpoch = isForkPostFulu(fork) ? proposalEpoch - 1 : proposalEpoch; + return Math.max(computeStartSlotAtEpoch(decisionEpoch) - 1, 0); } /** diff --git a/packages/validator/src/services/blockDuties.ts b/packages/validator/src/services/blockDuties.ts index 152c8935eb73..418240f8d86e 100644 --- a/packages/validator/src/services/blockDuties.ts +++ b/packages/validator/src/services/blockDuties.ts @@ -1,18 +1,28 @@ import {ApiClient, routes} from "@lodestar/api"; import {ChainForkConfig} from "@lodestar/config"; +import {isForkPostFulu} from "@lodestar/params"; import {computeEpochAtSlot, computeStartSlotAtEpoch} from "@lodestar/state-transition"; import {BLSPubkey, Epoch, RootHex, Slot} from "@lodestar/types"; import {sleep, toPubkeyHex} from "@lodestar/utils"; import {Metrics} from "../metrics.js"; import {PubkeyHex} from "../types.js"; -import {IClock, LoggerVc, differenceHex} from "../util/index.js"; +import {IClock, LoggerVc} from "../util/index.js"; +import {ChainHeaderTracker, HeadEventData} from "./chainHeaderTracker.js"; import {ValidatorStore} from "./validatorStore.js"; -/** This polls block duties 1s before the next epoch */ +/** + * Pre-Fulu only: poll next-epoch proposer duties ~1s before the boundary. Post-Fulu the 1-epoch + * deterministic lookahead lets us pre-fetch via `runEveryEpoch`, so this fast path is unused. + * + * Historical context: starting Jul 2023 we poll 1s before the next epoch because + * `PrepareNextSlotScheduler` (BN-side) usually finishes the upcoming-epoch transition in ~3s, + * so the proposer-duties query at ~1s pre-boundary lands on a hot cache. See: + * - https://github.com/ChainSafe/lodestar/issues/5792 + */ // TODO: change to 8333 (5/6 of slot) to do it 2s before the next epoch // once we have some improvement on epoch transition time // see https://github.com/ChainSafe/lodestar/issues/5792#issuecomment-1647457442 -// TODO GLOAS: re-evaluate timing +// TODO GLOAS: re-evaluate timing — Gloas may want the offset *after* the boundary const BLOCK_DUTIES_LOOKAHEAD_BPS = 9167; /** Only retain `HISTORICAL_DUTIES_EPOCHS` duties prior to the current epoch */ const HISTORICAL_DUTIES_EPOCHS = 2; @@ -30,18 +40,32 @@ export class BlockDutiesService { proposals for any validators which are not registered locally. */ private readonly proposers = new Map(); + /** + * Tracks which proposer pubkeys we have already notified for the active slot so that + * a late-arriving cache update (SSE-driven refetch, slow initial poll) only fires + * `notifyBlockProductionFn` for *newly discovered* proposers, never duplicates. + */ + private notifiedSlot: Slot = -1; + private readonly notifiedProposers = new Set(); + /** + * True once `notifyProposersForSlot` has been invoked for `notifiedSlot`, regardless of + * whether anything was notified. Any subsequent invocation that finds *new* proposers is + * therefore a late detection — the signal tracked by `newProposalDutiesDetected`. + */ + private notifiedSlotInitialPass = false; + constructor( private readonly config: ChainForkConfig, private readonly logger: LoggerVc, private readonly api: ApiClient, private readonly clock: IClock, private readonly validatorStore: ValidatorStore, + chainHeaderTracker: ChainHeaderTracker, private readonly metrics: Metrics | null ) { - // TODO: Instead of polling every CLOCK_SLOT, poll every CLOCK_EPOCH and track re-org events - // only then re-fetch the block duties. Make sure most clients (including Lodestar) - // properly emit the re-org event - clock.runEverySlot(this.runBlockDutiesTask); + clock.runEveryEpoch(this.runEveryEpochTask); + clock.runEverySlot(this.runEverySlotTask); + chainHeaderTracker.runOnNewHead(this.onNewHead); if (metrics) { metrics.proposerDutiesEpochCount.addCollect(() => { @@ -98,108 +122,191 @@ export class BlockDutiesService { } } - private runBlockDutiesTask = async (slot: Slot, signal: AbortSignal): Promise => { + /** + * Baseline per-epoch fetch. Fires at epoch boundaries (and once at startup). Post-Fulu the + * deterministic 1-epoch lookahead lets us also pre-fetch `epoch + 1`; pre-Fulu the next + * epoch's dep_root only stabilizes at the boundary and is handled by `runEverySlotTask`. + * + * Mid-epoch refreshes (e.g. reorgs) are driven by `onNewHead` instead of polling every slot. + */ + private runEveryEpochTask = async (epoch: Epoch): Promise => { try { - if (slot < 0) { - // Before genesis, fetch the genesis duties but don't notify block production - // Only fetch duties once since there is not possible to re-org. TODO: Review + if (epoch < GENESIS_EPOCH) { + // Pre-genesis: prime the genesis-epoch duties exactly once so the slot-0 proposer + // doesn't have to wait on a cold cache. Only fetch once since a pre-genesis re-org + // is not possible. TODO: Review. if (!this.proposers.has(GENESIS_EPOCH)) { await this.pollBeaconProposers(GENESIS_EPOCH); } - } else { - await this.pollBeaconProposersAndNotify(slot, signal); + return; + } + + await this.pollBeaconProposers(epoch); + + const nextEpoch = epoch + 1; + if (isForkPostFulu(this.config.getForkName(computeStartSlotAtEpoch(nextEpoch)))) { + await this.pollBeaconProposers(nextEpoch); } } catch (e) { - this.logger.error("Error on pollBeaconProposers", {}, e as Error); + this.logger.error("Error on runEveryEpochTask", {epoch}, e as Error); } finally { - this.pruneOldDuties(computeEpochAtSlot(slot)); + this.pruneOldDuties(Math.max(epoch, GENESIS_EPOCH)); } }; /** - * Download the proposer duties for the current epoch and store them in `this.proposers`. - * If there are any proposer for this slot, send out a notification to the block proposers. - * - * ## Note - * - * This function will potentially send *two* notifications to the `BlockService`; it will send a - * notification initially, then it will download the latest duties and send a *second* notification - * if those duties have changed. This behaviour simultaneously achieves the following: + * Slot-tick handler. Notifies block production for cached proposers in this slot, and on + * the last slot of a pre-Fulu epoch schedules the boundary fetch for `nextEpoch` duties. + * Reorg detection is handled by `onNewHead`, so this task does not re-poll on every slot. + */ + private runEverySlotTask = async (slot: Slot, signal: AbortSignal): Promise => { + try { + if (slot < GENESIS_SLOT) { + return; + } + + this.notifyProposersForSlot(slot); + + const nextEpoch = computeEpochAtSlot(slot) + 1; + const isLastSlotOfEpoch = computeStartSlotAtEpoch(nextEpoch) === slot + 1; + if (isLastSlotOfEpoch && !isForkPostFulu(this.config.getForkName(slot + 1))) { + // Pre-Fulu: 0-epoch proposer lookahead, so the next-epoch dep_root only becomes stable + // as the last block of the current epoch lands. Sleep until ~1s before the boundary + // then fetch — same timing as before this refactor. + this.pollBeaconProposersBeforeBoundary(slot, nextEpoch, signal).catch((e) => { + this.logger.error("Error on pollBeaconProposersBeforeBoundary", {nextEpoch}, e); + }); + } + } catch (e) { + this.logger.error("Error on runEverySlotTask", {slot}, e as Error); + } + }; + + /** + * SSE head-event handler. The beacon-API `head` event carries attester-duty dep_roots, + * which coincide with the proposer dep_roots at a fork-dependent offset: * - * 1. Block production can happen immediately and does not have to wait for the proposer duties to - * download. - * 2. We won't miss a block if the duties for the current slot happen to change with this poll. + * Pre-Fulu (proposer dep_root(E) = block@startSlot(E) - 1): + * currentDutyDependentRoot ≡ proposer_dep_root(currentEpoch) + * (next-epoch proposer dep_root is not exposed; pre-Fulu falls back to the + * `runEverySlotTask` boundary poll.) * - * This sounds great, but is it safe? Firstly, the additional notification will only contain block - * producers that were not included in the first notification. This should be safety enough. - * However, we also have the slashing protection as a second line of defense. These two factors - * provide an acceptable level of safety. + * Post-Fulu (proposer dep_root(E) = block@startSlot(E - 1) - 1, EIP-7917): + * previousDutyDependentRoot ≡ proposer_dep_root(currentEpoch) + * currentDutyDependentRoot ≡ proposer_dep_root(nextEpoch) * - * It's important to note that since there is a 0-epoch look-ahead (i.e., no look-ahead) for block - * proposers then it's very likely that a proposal for the first slot of the epoch will need go - * through the slow path every time. I.e., the proposal will only happen after we've been able to - * download and process the duties from the BN. This means it is very important to ensure this - * function is as fast as possible. - * - Starting from Jul 2023, we poll proposers 1s before the next epoch thanks to PrepareNextSlotScheduler - * usually finishes in 3s. + * On a dep_root mismatch (reorg, or initial sync delivering a fresher head) we refetch + * just the affected epoch, mirroring `AttestationDutiesService.onNewHead`. */ - private async pollBeaconProposersAndNotify(currentSlot: Slot, signal: AbortSignal): Promise { - const nextEpoch = computeEpochAtSlot(currentSlot) + 1; - const isLastSlotEpoch = computeStartSlotAtEpoch(nextEpoch) === currentSlot + 1; - if (isLastSlotEpoch) { - // no need to await for other steps, just poll proposers for next epoch - this.pollBeaconProposersNextEpoch(currentSlot, nextEpoch, signal).catch((e) => { - this.logger.error("Error on pollBeaconProposersNextEpoch", {}, e); - }); - } + private onNewHead = async ({ + slot, + previousDutyDependentRoot, + currentDutyDependentRoot, + }: HeadEventData): Promise => { + const currentEpoch = computeEpochAtSlot(slot); + const isPostFulu = isForkPostFulu(this.config.getForkName(slot)); - // Notify the block proposal service for any proposals that we have in our cache. - const initialBlockProposers = this.getblockProposersAtSlot(currentSlot); - if (initialBlockProposers.length > 0) { - this.notifyBlockProductionFn(currentSlot, initialBlockProposers); + if (isPostFulu) { + await this.refetchIfDepRootChanged(currentEpoch, previousDutyDependentRoot); + await this.refetchIfDepRootChanged(currentEpoch + 1, currentDutyDependentRoot); + } else { + await this.refetchIfDepRootChanged(currentEpoch, currentDutyDependentRoot); } + }; - // Poll proposers again for the same slot - await this.pollBeaconProposers(computeEpochAtSlot(currentSlot)); - - // Compute the block proposers for this slot again, now that we've received an update from the BN. - // - // Then, compute the difference between these two sets to obtain a set of block proposers - // which were not included in the initial notification to the `BlockService`. - const newBlockProducers = this.getblockProposersAtSlot(currentSlot); - const additionalBlockProducers = differenceHex(initialBlockProposers, newBlockProducers); - - // If there are any new proposers for this slot, send a notification so they produce a block. - // - // See the function-level documentation for more reasoning about this behaviour. - if (additionalBlockProducers.length > 0) { - this.notifyBlockProductionFn(currentSlot, additionalBlockProducers); - this.logger.debug("Detected new block proposer", {currentSlot}); - this.metrics?.newProposalDutiesDetected.inc(); + private async refetchIfDepRootChanged(epoch: Epoch, expectedDepRoot: RootHex): Promise { + const cached = this.proposers.get(epoch); + if (!cached || cached.dependentRoot === expectedDepRoot) { + return; } + + this.logger.debug("Proposer duties dep_root changed, refetching", { + epoch, + priorDependentRoot: cached.dependentRoot, + newDependentRoot: expectedDepRoot, + }); + await this.pollBeaconProposers(epoch); } /** - * This is to avoid some delay on the first slot of the epoch when validators have proposal duties. - * See https://github.com/ChainSafe/lodestar/issues/5792 + * Pre-Fulu boundary fetch. Because pre-Fulu proposer shuffling has 0-epoch look-ahead, a + * proposal for the first slot of the new epoch otherwise goes through the slow path every + * time: the proposal can only happen *after* we download and process the new duties from + * the BN. Polling ~1s before the boundary, while `PrepareNextSlotScheduler` is finishing + * the upcoming-epoch transition, lets us land on a hot BN cache and avoid the miss. + * + * See https://github.com/ChainSafe/lodestar/issues/5792. */ - private async pollBeaconProposersNextEpoch(currentSlot: Slot, nextEpoch: Epoch, signal: AbortSignal): Promise { + private async pollBeaconProposersBeforeBoundary( + currentSlot: Slot, + nextEpoch: Epoch, + signal: AbortSignal + ): Promise { const nextSlot = currentSlot + 1; const lookAheadMs = this.config.SLOT_DURATION_MS - this.config.getSlotComponentDurationMs(BLOCK_DUTIES_LOOKAHEAD_BPS); await sleep(this.clock.msToSlot(nextSlot) - lookAheadMs, signal); - this.logger.debug("Polling proposers for next epoch", {nextEpoch, nextSlot}); - // Poll proposers for the next epoch + this.logger.debug("Polling proposers for the next epoch", {nextEpoch, currentSlot}); await this.pollBeaconProposers(nextEpoch); } + /** + * Notify block production for *newly discovered* proposers in this slot. Notifications are + * deduplicated per-slot so that a late SSE refetch can extend the proposer set without + * triggering a duplicate `createAndPublishBlock` for already-notified validators. + * + * ## Multi-notification safety + * + * Within a single slot the cache can be updated from several sources (cold-cache backfill at + * startup, SSE-driven reorg refetch). Each update may fire this function again. The contract + * we keep is: each pubkey is notified *at most once per slot*. The additional notifications + * only carry proposers that were not part of an earlier notification. + * + * Is this safe? Firstly, the dedup above guarantees we never ask the same validator to + * propose twice for the same slot. Secondly, slashing protection in `ValidatorStore` acts as + * a second line of defense should the dedup ever fail. Together they provide an acceptable + * level of safety for the "notify-from-cache, refine-after-refetch" pattern. + */ + private notifyProposersForSlot(slot: Slot): void { + if (slot !== this.notifiedSlot) { + this.notifiedSlot = slot; + this.notifiedSlotInitialPass = false; + this.notifiedProposers.clear(); + } + + const isLateDetection = this.notifiedSlotInitialPass; + this.notifiedSlotInitialPass = true; + + const newProposers: BLSPubkey[] = []; + for (const pubkey of this.getblockProposersAtSlot(slot)) { + const pubkeyHex = toPubkeyHex(pubkey); + if (!this.notifiedProposers.has(pubkeyHex)) { + this.notifiedProposers.add(pubkeyHex); + newProposers.push(pubkey); + } + } + + if (newProposers.length === 0) { + return; + } + + if (isLateDetection) { + this.metrics?.newProposalDutiesDetected.inc(); + this.logger.debug("Detected new block proposer", {slot}); + } + this.notifyBlockProductionFn(slot, newProposers); + } + private async pollBeaconProposers(epoch: Epoch): Promise { // Only download duties and push out additional block production events if we have some validators. if (!this.validatorStore.hasSomeValidators()) { return; } - const res = await this.api.validator.getProposerDuties({epoch}); + // Post-Fulu the proposer dependent root changed (deterministic proposer lookahead) + const res = isForkPostFulu(this.config.getForkName(computeStartSlotAtEpoch(epoch))) + ? await this.api.validator.getProposerDutiesV2({epoch}) + : await this.api.validator.getProposerDuties({epoch}); const proposerDuties = res.value(); const {dependentRoot} = res.meta(); const relevantDuties = proposerDuties.filter((duty) => { @@ -210,6 +317,11 @@ export class BlockDutiesService { this.logger.debug("Downloaded proposer duties", {epoch, dependentRoot, count: relevantDuties.length}); const prior = this.proposers.get(epoch); + // Concurrent polls for the same epoch (e.g. `onNewHead` and `runEveryEpochTask` racing) + // both write here last-write-wins. The pre-refactor per-slot poll healed any stale write + // on the next slot; in the event-driven model staleness can persist until the next + // dep_root change. In practice the same BN serves both calls so they return identical + // payloads — accept the rare race rather than serialising fetches. this.proposers.set(epoch, {dependentRoot, data: relevantDuties}); if (prior && prior.dependentRoot !== dependentRoot) { @@ -219,6 +331,12 @@ export class BlockDutiesService { dependentRoot, }); } + + // If this fetch revealed proposer(s) for the active slot that the last `runEverySlotTask` + // missed (cold cache at startup, or duties shifted by a reorg), notify now. + if (this.notifiedSlot >= GENESIS_SLOT && computeEpochAtSlot(this.notifiedSlot) === epoch) { + this.notifyProposersForSlot(this.notifiedSlot); + } } /** Run once per epoch to prune `this.proposers` map */ diff --git a/packages/validator/src/validator.ts b/packages/validator/src/validator.ts index 97e4f36be3ab..57a2c35da44c 100644 --- a/packages/validator/src/validator.ts +++ b/packages/validator/src/validator.ts @@ -235,7 +235,15 @@ export class Validator { const chainHeaderTracker = new ChainHeaderTracker(config, logger, api, emitter); const syncingStatusTracker = new SyncingStatusTracker(logger, api, clock, metrics); - const blockDutiesService = new BlockDutiesService(config, loggerVc, api, clock, validatorStore, metrics); + const blockDutiesService = new BlockDutiesService( + config, + loggerVc, + api, + clock, + validatorStore, + chainHeaderTracker, + metrics + ); const blockProposingService = new BlockProposingService( config, diff --git a/packages/validator/test/unit/services/block.test.ts b/packages/validator/test/unit/services/block.test.ts index 71e46620cdd9..cb23e43a0fb6 100644 --- a/packages/validator/test/unit/services/block.test.ts +++ b/packages/validator/test/unit/services/block.test.ts @@ -1,4 +1,4 @@ -import {afterEach, beforeEach, describe, expect, it, vi} from "vitest"; +import {Mocked, afterEach, beforeEach, describe, expect, it, vi} from "vitest"; import {SecretKey} from "@chainsafe/blst"; import {toHexString} from "@chainsafe/ssz"; import {routes} from "@lodestar/api"; @@ -9,6 +9,7 @@ import {ProducedBlockSource, ssz} from "@lodestar/types"; import {sleep} from "@lodestar/utils"; import {BlockProposingService} from "../../../src/services/block.js"; import {BlockDutiesService} from "../../../src/services/blockDuties.js"; +import {ChainHeaderTracker} from "../../../src/services/chainHeaderTracker.js"; import {ValidatorStore} from "../../../src/services/validatorStore.js"; import {getApiClientStub, mockApiResponse} from "../../utils/apiStub.js"; import {ClockMock} from "../../utils/clock.js"; @@ -16,6 +17,7 @@ import {loggerVc} from "../../utils/logger.js"; import {ZERO_HASH_HEX} from "../../utils/types.js"; vi.mock("../../../src/services/validatorStore.js"); +vi.mock("../../../src/services/chainHeaderTracker.js"); describe("BlockDutiesService", () => { const api = getApiClientStub(); @@ -24,6 +26,8 @@ describe("BlockDutiesService", () => { let pubkeys: Uint8Array[]; // Initialize pubkeys in before() so bls is already initialized const config = createChainForkConfig(mainnetConfig); + // @ts-expect-error - Mocked class don't need parameters + const chainHeaderTracker = new ChainHeaderTracker() as Mocked; let controller: AbortController; // To stop clock beforeEach(() => { @@ -55,7 +59,15 @@ describe("BlockDutiesService", () => { ); const clock = new ClockMock(); - const dutiesService = new BlockDutiesService(config, loggerVc, api, clock, validatorStore, null); + const dutiesService = new BlockDutiesService( + config, + loggerVc, + api, + clock, + validatorStore, + chainHeaderTracker, + null + ); const blockService = new BlockProposingService(config, loggerVc, api, clock, validatorStore, dutiesService, null, { broadcastValidation: routes.beacon.BroadcastValidation.consensus, blindedLocal: false, @@ -129,7 +141,15 @@ describe("BlockDutiesService", () => { ); const clock = new ClockMock(); - const dutiesService = new BlockDutiesService(config, loggerVc, api, clock, validatorStore, null); + const dutiesService = new BlockDutiesService( + config, + loggerVc, + api, + clock, + validatorStore, + chainHeaderTracker, + null + ); const blockService = new BlockProposingService(config, loggerVc, api, clock, validatorStore, dutiesService, null, { broadcastValidation: routes.beacon.BroadcastValidation.consensus, blindedLocal: true, diff --git a/packages/validator/test/unit/services/blockDuties.test.ts b/packages/validator/test/unit/services/blockDuties.test.ts index e09e73efa36c..89406059c05a 100644 --- a/packages/validator/test/unit/services/blockDuties.test.ts +++ b/packages/validator/test/unit/services/blockDuties.test.ts @@ -1,12 +1,15 @@ import {toBufferBE} from "@vekexasia/bigint-buffer2"; -import {afterEach, beforeAll, beforeEach, describe, expect, it, vi} from "vitest"; +import {Mocked, afterEach, beforeAll, beforeEach, describe, expect, it, vi} from "vitest"; import {SecretKey} from "@chainsafe/blst"; import {toHexString} from "@chainsafe/ssz"; import {routes} from "@lodestar/api"; import {createChainForkConfig} from "@lodestar/config"; import {config as defaultConfig} from "@lodestar/config/default"; +import {getConfig} from "@lodestar/config/test-utils"; +import {ForkName} from "@lodestar/params"; import {toHex} from "@lodestar/utils"; import {BlockDutiesService} from "../../../src/services/blockDuties.js"; +import {ChainHeaderTracker, HeadEventData} from "../../../src/services/chainHeaderTracker.js"; import {ValidatorStore} from "../../../src/services/validatorStore.js"; import {getApiClientStub, mockApiResponse} from "../../utils/apiStub.js"; import {ClockMock} from "../../utils/clock.js"; @@ -14,96 +17,331 @@ import {loggerVc} from "../../utils/logger.js"; import {ZERO_HASH_HEX} from "../../utils/types.js"; import {initValidatorStore} from "../../utils/validatorStore.js"; +vi.mock("../../../src/services/chainHeaderTracker.js"); + describe("BlockDutiesService", () => { const api = getApiClientStub(); - const config = createChainForkConfig(defaultConfig); + const preFuluConfig = createChainForkConfig(defaultConfig); + const postFuluConfig = getConfig(ForkName.fulu); let validatorStore: ValidatorStore; let pubkeys: Uint8Array[]; // Initialize pubkeys in before() so bls is already initialized + // @ts-expect-error - Mocked class don't need parameters + const chainHeaderTracker = new ChainHeaderTracker() as Mocked; + beforeAll(async () => { const secretKeys = Array.from({length: 3}, (_, i) => SecretKey.fromBytes(toBufferBE(BigInt(i + 1), 32))); pubkeys = secretKeys.map((sk) => sk.toPublicKey().toBytes()); validatorStore = await initValidatorStore(secretKeys, api); + + // Defensive default for the post-Fulu v2 branch — individual tests override as needed. + api.validator.getProposerDutiesV2.mockResolvedValue( + mockApiResponse({data: [], meta: {dependentRoot: ZERO_HASH_HEX, executionOptimistic: false}}) + ); }); let controller: AbortController; // To stop clock + let onNewHeadCallback: (headEvent: HeadEventData) => Promise; + beforeEach(() => { controller = new AbortController(); + vi.spyOn(chainHeaderTracker, "runOnNewHead"); + chainHeaderTracker.runOnNewHead.mockImplementation((callback) => { + onNewHeadCallback = callback; + }); }); afterEach(() => controller.abort()); - it("Should fetch and persist block duties", async () => { - // Reply with some duties - const slot = 0; // genesisTime is right now, so test with slot = currentSlot - const duties: routes.validator.ProposerDutyList = [{slot: slot, validatorIndex: 0, pubkey: pubkeys[0]}]; + it("Should fetch and persist block duties on epoch tick, notify on slot tick", async () => { + const slot = 0; + const duties: routes.validator.ProposerDutyList = [{slot, validatorIndex: 0, pubkey: pubkeys[0]}]; api.validator.getProposerDuties.mockResolvedValue( mockApiResponse({data: duties, meta: {dependentRoot: ZERO_HASH_HEX, executionOptimistic: false}}) ); - const notifyBlockProductionFn = vi.fn(); // Returns void - + const notifyBlockProductionFn = vi.fn(); const clock = new ClockMock(); - const dutiesService = new BlockDutiesService(config, loggerVc, api, clock, validatorStore, null); + const dutiesService = new BlockDutiesService( + preFuluConfig, + loggerVc, + api, + clock, + validatorStore, + chainHeaderTracker, + null + ); dutiesService.setNotifyBlockProductionFn(notifyBlockProductionFn); - // Trigger clock onSlot for slot 0 - await clock.tickSlotFns(0, controller.signal); + // Epoch tick populates the cache, slot tick fires the notification + await clock.tickEpochFns(0, controller.signal); + await clock.tickSlotFns(slot, controller.signal); - // Duties for this epoch should be persisted expect(Object.fromEntries(dutiesService["proposers"])).toEqual({0: {dependentRoot: ZERO_HASH_HEX, data: duties}}); - expect(dutiesService.getblockProposersAtSlot(slot)).toEqual([pubkeys[0]]); - expect(notifyBlockProductionFn).toHaveBeenCalledOnce(); + expect(notifyBlockProductionFn).toHaveBeenCalledWith(slot, [pubkeys[0]]); }); - it("Should call notifyBlockProductionFn again on duties re-org", async () => { - // A re-org will happen at slot 1 - const dependentRootDiff = toHex(Buffer.alloc(32, 1)); - const dutiesBeforeReorg: routes.validator.ProposerDutyList = [{slot: 1, validatorIndex: 0, pubkey: pubkeys[0]}]; - const dutiesAfterReorg: routes.validator.ProposerDutyList = [{slot: 1, validatorIndex: 1, pubkey: pubkeys[1]}]; + it("Should notify cached proposers even if slot tick precedes epoch tick (cold-cache startup)", async () => { + const slot = 0; + const duties: routes.validator.ProposerDutyList = [{slot, validatorIndex: 0, pubkey: pubkeys[0]}]; - const notifyBlockProductionFn = vi.fn(); // Returns void + api.validator.getProposerDuties.mockResolvedValue( + mockApiResponse({data: duties, meta: {dependentRoot: ZERO_HASH_HEX, executionOptimistic: false}}) + ); - // Clock will call runAttesterDutiesTasks() immediately + const notifyBlockProductionFn = vi.fn(); const clock = new ClockMock(); - const dutiesService = new BlockDutiesService(config, loggerVc, api, clock, validatorStore, null); + const dutiesService = new BlockDutiesService( + preFuluConfig, + loggerVc, + api, + clock, + validatorStore, + chainHeaderTracker, + null + ); dutiesService.setNotifyBlockProductionFn(notifyBlockProductionFn); - // Trigger clock onSlot for slot 0 + // Slot ticks first (cache empty → no notify); then epoch tick fetches and back-fills the + // notification for the active slot. + await clock.tickSlotFns(slot, controller.signal); + expect(notifyBlockProductionFn).not.toHaveBeenCalled(); + + await clock.tickEpochFns(0, controller.signal); + expect(notifyBlockProductionFn).toHaveBeenCalledOnce(); + expect(notifyBlockProductionFn).toHaveBeenCalledWith(slot, [pubkeys[0]]); + }); + + it("Post-Fulu epoch tick fetches current and next epoch proposer duties", async () => { + const dutiesEpoch0: routes.validator.ProposerDutyList = [{slot: 0, validatorIndex: 0, pubkey: pubkeys[0]}]; + const dutiesEpoch1: routes.validator.ProposerDutyList = [{slot: 32, validatorIndex: 1, pubkey: pubkeys[1]}]; + const depRootEpoch0 = ZERO_HASH_HEX; + const depRootEpoch1 = toHex(Buffer.alloc(32, 9)); + + api.validator.getProposerDutiesV2.mockImplementation(async ({epoch}) => + epoch === 0 + ? mockApiResponse({data: dutiesEpoch0, meta: {dependentRoot: depRootEpoch0, executionOptimistic: false}}) + : mockApiResponse({data: dutiesEpoch1, meta: {dependentRoot: depRootEpoch1, executionOptimistic: false}}) + ); + + const clock = new ClockMock(); + const dutiesService = new BlockDutiesService( + postFuluConfig, + loggerVc, + api, + clock, + validatorStore, + chainHeaderTracker, + null + ); + + await clock.tickEpochFns(0, controller.signal); + + expect(api.validator.getProposerDutiesV2).toHaveBeenCalledTimes(2); + expect(api.validator.getProposerDutiesV2.mock.calls.map((c) => c[0].epoch)).toEqual([0, 1]); + expect(Object.fromEntries(dutiesService["proposers"])).toEqual({ + 0: {dependentRoot: depRootEpoch0, data: dutiesEpoch0}, + 1: {dependentRoot: depRootEpoch1, data: dutiesEpoch1}, + }); + }); + + it("Pre-Fulu epoch tick fetches only current epoch", async () => { + api.validator.getProposerDuties.mockResolvedValue( + mockApiResponse({data: [], meta: {dependentRoot: ZERO_HASH_HEX, executionOptimistic: false}}) + ); + + const clock = new ClockMock(); + new BlockDutiesService(preFuluConfig, loggerVc, api, clock, validatorStore, chainHeaderTracker, null); + + await clock.tickEpochFns(0, controller.signal); + + expect(api.validator.getProposerDuties).toHaveBeenCalledTimes(1); + expect(api.validator.getProposerDuties.mock.calls[0][0]).toEqual({epoch: 0}); + }); + + it("Refetches and re-notifies on head-event dep_root mismatch (reorg)", async () => { + const slot = 1; + const dependentRootDiff = toHex(Buffer.alloc(32, 1)); + const dutiesBeforeReorg: routes.validator.ProposerDutyList = [{slot, validatorIndex: 0, pubkey: pubkeys[0]}]; + const dutiesAfterReorg: routes.validator.ProposerDutyList = [{slot, validatorIndex: 1, pubkey: pubkeys[1]}]; + api.validator.getProposerDuties.mockResolvedValue( mockApiResponse({data: dutiesBeforeReorg, meta: {dependentRoot: ZERO_HASH_HEX, executionOptimistic: false}}) ); - await clock.tickSlotFns(0, controller.signal); - // Trigger clock onSlot for slot 1 - Return different duties for slot 1 + const notifyBlockProductionFn = vi.fn(); + const clock = new ClockMock(); + const dutiesService = new BlockDutiesService( + preFuluConfig, + loggerVc, + api, + clock, + validatorStore, + chainHeaderTracker, + null + ); + dutiesService.setNotifyBlockProductionFn(notifyBlockProductionFn); + + await clock.tickEpochFns(0, controller.signal); + await clock.tickSlotFns(slot, controller.signal); + + expect(notifyBlockProductionFn).toHaveBeenCalledOnce(); + expect(notifyBlockProductionFn.mock.calls[0]).toEqual([slot, [pubkeys[0]]]); + + // Simulate SSE: new head with a different dep_root for the current epoch api.validator.getProposerDuties.mockResolvedValue( mockApiResponse({data: dutiesAfterReorg, meta: {dependentRoot: dependentRootDiff, executionOptimistic: false}}) ); - await clock.tickSlotFns(1, controller.signal); + await onNewHeadCallback({ + slot, + head: dependentRootDiff, + previousDutyDependentRoot: ZERO_HASH_HEX, + currentDutyDependentRoot: dependentRootDiff, + }); - // Should persist the dutiesAfterReorg expect(Object.fromEntries(dutiesService["proposers"])).toEqual({ 0: {dependentRoot: dependentRootDiff, data: dutiesAfterReorg}, }); + expect(notifyBlockProductionFn).toHaveBeenCalledTimes(2); + expect(notifyBlockProductionFn.mock.calls[1]).toEqual([slot, [pubkeys[1]]]); + }); - expect(notifyBlockProductionFn).toBeCalledTimes(2); + it("Does not refetch when head-event dep_root matches the cache", async () => { + const slot = 1; + const duties: routes.validator.ProposerDutyList = [{slot, validatorIndex: 0, pubkey: pubkeys[0]}]; - expect(notifyBlockProductionFn.mock.calls[0]).toEqual([1, [pubkeys[0]]]); - expect(notifyBlockProductionFn.mock.calls[1]).toEqual([1, [pubkeys[1]]]); + api.validator.getProposerDuties.mockResolvedValue( + mockApiResponse({data: duties, meta: {dependentRoot: ZERO_HASH_HEX, executionOptimistic: false}}) + ); + + const clock = new ClockMock(); + new BlockDutiesService(preFuluConfig, loggerVc, api, clock, validatorStore, chainHeaderTracker, null); + + await clock.tickEpochFns(0, controller.signal); + expect(api.validator.getProposerDuties).toHaveBeenCalledTimes(1); + + await onNewHeadCallback({ + slot, + head: ZERO_HASH_HEX, + previousDutyDependentRoot: ZERO_HASH_HEX, + currentDutyDependentRoot: ZERO_HASH_HEX, + }); + + // No refetch because dep_root matched + expect(api.validator.getProposerDuties).toHaveBeenCalledTimes(1); + }); + + it("Post-Fulu head event detects dep_root change for both current and next epoch", async () => { + const depRootEpoch0 = toHex(Buffer.alloc(32, 10)); + const depRootEpoch1 = toHex(Buffer.alloc(32, 11)); + const depRootEpoch0Reorg = toHex(Buffer.alloc(32, 20)); + const depRootEpoch1Reorg = toHex(Buffer.alloc(32, 21)); + + let epoch0DepRoot = depRootEpoch0; + let epoch1DepRoot = depRootEpoch1; + api.validator.getProposerDutiesV2.mockImplementation(async ({epoch}) => + mockApiResponse({ + data: [], + meta: {dependentRoot: epoch === 0 ? epoch0DepRoot : epoch1DepRoot, executionOptimistic: false}, + }) + ); + + const clock = new ClockMock(); + new BlockDutiesService(postFuluConfig, loggerVc, api, clock, validatorStore, chainHeaderTracker, null); + + await clock.tickEpochFns(0, controller.signal); + expect(api.validator.getProposerDutiesV2).toHaveBeenCalledTimes(2); + + // Reorg invalidates both epochs + epoch0DepRoot = depRootEpoch0Reorg; + epoch1DepRoot = depRootEpoch1Reorg; + await onNewHeadCallback({ + slot: 0, + head: ZERO_HASH_HEX, + previousDutyDependentRoot: depRootEpoch0Reorg, // post-Fulu maps to proposer_dep_root(currentEpoch) + currentDutyDependentRoot: depRootEpoch1Reorg, // post-Fulu maps to proposer_dep_root(nextEpoch) + }); + + expect(api.validator.getProposerDutiesV2).toHaveBeenCalledTimes(4); + // Each epoch refetched once + const refetchEpochs = api.validator.getProposerDutiesV2.mock.calls.slice(2).map((c) => c[0].epoch); + expect(refetchEpochs.sort()).toEqual([0, 1]); + }); + + it("Pre-Fulu last slot of epoch schedules a boundary fetch for nextEpoch", async () => { + const epoch0Duties: routes.validator.ProposerDutyList = []; + const epoch1Duties: routes.validator.ProposerDutyList = [{slot: 32, validatorIndex: 0, pubkey: pubkeys[0]}]; + + api.validator.getProposerDuties.mockImplementation(async ({epoch}) => + mockApiResponse({ + data: epoch === 0 ? epoch0Duties : epoch1Duties, + meta: {dependentRoot: ZERO_HASH_HEX, executionOptimistic: false}, + }) + ); + + const clock = new ClockMock(); + new BlockDutiesService(preFuluConfig, loggerVc, api, clock, validatorStore, chainHeaderTracker, null); + + await clock.tickEpochFns(0, controller.signal); + expect(api.validator.getProposerDuties).toHaveBeenCalledTimes(1); + expect(api.validator.getProposerDuties.mock.calls[0][0]).toEqual({epoch: 0}); + + // Last slot of epoch 0 → schedules `pollBeaconProposersBeforeBoundary` (fire-and-forget). + // ClockMock returns 0 for `msToSlot`, so the sleep resolves immediately. + await clock.tickSlotFns(31, controller.signal); + // Yield to let the fire-and-forget poll resolve + await new Promise((resolve) => setTimeout(resolve, 10)); + + expect(api.validator.getProposerDuties).toHaveBeenCalledTimes(2); + expect(api.validator.getProposerDuties.mock.calls[1][0]).toEqual({epoch: 1}); + }); + + it("Pre-Fulu mid-epoch slot tick does NOT schedule a boundary fetch", async () => { + api.validator.getProposerDuties.mockResolvedValue( + mockApiResponse({data: [], meta: {dependentRoot: ZERO_HASH_HEX, executionOptimistic: false}}) + ); + + const clock = new ClockMock(); + new BlockDutiesService(preFuluConfig, loggerVc, api, clock, validatorStore, chainHeaderTracker, null); + + await clock.tickEpochFns(0, controller.signal); + expect(api.validator.getProposerDuties).toHaveBeenCalledTimes(1); + + // Mid-epoch slot tick: notify only, no boundary fetch + await clock.tickSlotFns(15, controller.signal); + await new Promise((resolve) => setTimeout(resolve, 10)); + + expect(api.validator.getProposerDuties).toHaveBeenCalledTimes(1); + }); + + it("Post-Fulu last slot of epoch does NOT schedule a pre-Fulu boundary fetch", async () => { + api.validator.getProposerDutiesV2.mockResolvedValue( + mockApiResponse({data: [], meta: {dependentRoot: ZERO_HASH_HEX, executionOptimistic: false}}) + ); + + const clock = new ClockMock(); + new BlockDutiesService(postFuluConfig, loggerVc, api, clock, validatorStore, chainHeaderTracker, null); + + await clock.tickEpochFns(0, controller.signal); + const callsAfterEpochTick = api.validator.getProposerDutiesV2.mock.calls.length; + + await clock.tickSlotFns(31, controller.signal); + await new Promise((resolve) => setTimeout(resolve, 10)); + + // No extra fetch beyond what `runEveryEpoch` already did — boundary poll is pre-Fulu only + expect(api.validator.getProposerDutiesV2.mock.calls.length).toBe(callsAfterEpochTick); }); - it("Should remove signer from duty", async () => { - // Reply with some duties - const slot = 0; // genesisTime is right now, so test with slot = currentSlot + it("Should remove signer from duty across epochs", async () => { const duties: routes.validator.ProposerDutyList = [ - {slot: slot, validatorIndex: 0, pubkey: pubkeys[0]}, - {slot: slot, validatorIndex: 1, pubkey: pubkeys[1]}, + {slot: 0, validatorIndex: 0, pubkey: pubkeys[0]}, + {slot: 0, validatorIndex: 1, pubkey: pubkeys[1]}, {slot: 33, validatorIndex: 2, pubkey: pubkeys[2]}, ]; const dutiesRemoved: routes.validator.ProposerDutyList = [ - {slot: slot, validatorIndex: 1, pubkey: pubkeys[1]}, + {slot: 0, validatorIndex: 1, pubkey: pubkeys[1]}, {slot: 33, validatorIndex: 2, pubkey: pubkeys[2]}, ]; @@ -111,26 +349,27 @@ describe("BlockDutiesService", () => { mockApiResponse({data: duties, meta: {dependentRoot: ZERO_HASH_HEX, executionOptimistic: false}}) ); - const notifyBlockProductionFn = vi.fn(); // Returns void - const clock = new ClockMock(); - const dutiesService = new BlockDutiesService(config, loggerVc, api, clock, validatorStore, null); - dutiesService.setNotifyBlockProductionFn(notifyBlockProductionFn); + const dutiesService = new BlockDutiesService( + preFuluConfig, + loggerVc, + api, + clock, + validatorStore, + chainHeaderTracker, + null + ); - // Trigger clock onSlot for slot 0 - await clock.tickSlotFns(0, controller.signal); - await clock.tickSlotFns(32, controller.signal); + await clock.tickEpochFns(0, controller.signal); + await clock.tickEpochFns(1, controller.signal); - // first confirm the duties for the epochs was persisted expect(Object.fromEntries(dutiesService["proposers"])).toEqual({ 0: {dependentRoot: ZERO_HASH_HEX, data: duties}, 1: {dependentRoot: ZERO_HASH_HEX, data: duties}, }); - // then remove a signers public key dutiesService.removeDutiesForKey(toHexString(pubkeys[0])); - // confirm that the duties no longer contain the signers public key expect(Object.fromEntries(dutiesService["proposers"])).toEqual({ 0: {dependentRoot: ZERO_HASH_HEX, data: dutiesRemoved}, 1: {dependentRoot: ZERO_HASH_HEX, data: dutiesRemoved}, diff --git a/packages/validator/test/utils/apiStub.ts b/packages/validator/test/utils/apiStub.ts index 303c04dc51e4..3375ee09e3a5 100644 --- a/packages/validator/test/utils/apiStub.ts +++ b/packages/validator/test/utils/apiStub.ts @@ -29,6 +29,7 @@ export function getApiClientStub(): ApiClientStub { }, validator: { getProposerDuties: vi.fn(), + getProposerDutiesV2: vi.fn(), getAttesterDuties: vi.fn(), getPtcDuties: vi.fn(), prepareBeaconCommitteeSubnet: vi.fn(),