diff --git a/cl/beacon/handler/block_production.go b/cl/beacon/handler/block_production.go
index 5533cf7b1ec..8fd483500d0 100644
--- a/cl/beacon/handler/block_production.go
+++ b/cl/beacon/handler/block_production.go
@@ -46,6 +46,7 @@ import (
"github.com/erigontech/erigon/cl/gossip"
"github.com/erigontech/erigon/cl/persistence/beacon_indicies"
"github.com/erigontech/erigon/cl/phase1/core/state"
+ "github.com/erigontech/erigon/cl/phase1/execution_client"
"github.com/erigontech/erigon/cl/phase1/forkchoice"
"github.com/erigontech/erigon/cl/phase1/network/subnets"
"github.com/erigontech/erigon/cl/pool"
@@ -208,16 +209,20 @@ func attestationDue(cfg *clparams.BeaconChainConfig, stateVersion clparams.State
return time.Duration(cfg.AttestationDueMs(stateVersion.AfterOrEqual(clparams.GloasVersion))) * time.Millisecond
}
-// computeBlockBuilderWindow returns when to first poll for the assembled payload and when to stop,
-// reserving a publication margin before the attestation deadline (see payloadPublicationDivisor).
-func computeBlockBuilderWindow(now, slotStart time.Time, cfg *clparams.BeaconChainConfig, stateVersion clparams.StateVersion) blockBuilderWindow {
+// computeBlockBuilderWindow uses the earlier collection window only for a sufficiently warmed
+// builder. Collecting stops the builder, so the earlier window trades the transactions that would
+// have arrived during the rest of the slot for the time it takes to sign, publish and gossip the
+// block before attesters vote on it.
+func computeBlockBuilderWindow(now, slotStart time.Time, cfg *clparams.BeaconChainConfig, stateVersion clparams.StateVersion, prepared bool) blockBuilderWindow {
due := attestationDue(cfg, stateVersion)
- grabBy := slotStart.Add(due - due/payloadPublicationDivisor)
- firstGetAt := grabBy.Add(-minPayloadPollingWindow)
+ pollUntil := slotStart.Add(unpreparedGrabOffset(due))
+ firstGetAt := pollUntil.Add(-minPayloadPollingWindow)
+ if prepared {
+ firstGetAt = slotStart.Add(preparedGrabOffset(due)).Add(-minPayloadPollingWindow)
+ }
if firstGetAt.Before(now) {
firstGetAt = now
}
- pollUntil := grabBy
if pollUntil.Before(firstGetAt) {
pollUntil = firstGetAt
}
@@ -227,6 +232,100 @@ func computeBlockBuilderWindow(now, slotStart time.Time, cfg *clparams.BeaconCha
}
}
+// unpreparedGrabOffset is when production starts polling for a payload it did not prime, and
+// preparedGrabOffset is when it may start for one it did. Their difference is the warm-up a
+// primed builder must already have, so the two paths give a builder the same total build time.
+func unpreparedGrabOffset(due time.Duration) time.Duration {
+ return due - due/payloadPublicationDivisor
+}
+
+func preparedGrabOffset(due time.Duration) time.Duration {
+ return due / payloadPublicationDivisor
+}
+
+func preparedPayloadMinimumAge(cfg *clparams.BeaconChainConfig, stateVersion clparams.StateVersion) time.Duration {
+ due := attestationDue(cfg, stateVersion)
+ return max(unpreparedGrabOffset(due)-preparedGrabOffset(due), 0)
+}
+
+const forkChoiceBusyRetryDelay = 100 * time.Millisecond
+
+// forkChoiceUpdateForProposal bounds acquiring a payload id by the moment the payload has to be
+// collected, so neither the retries here nor the assemble retry below them can eat the margin the
+// block needs to reach attesters. One attempt always runs past that bound, because failing a
+// proposal outright is worse than answering late. The head is not re-read between attempts: a
+// proposal is committed to the parent it is being built on.
+func (a *ApiHandler) forkChoiceUpdateForProposal(
+ ctx context.Context,
+ targetSlot uint64,
+ finalized, safe, head common.Hash,
+ attrs *engine_types.PayloadAttributes,
+ stateVersion clparams.StateVersion,
+) ([]byte, error) {
+ collectAt := a.ethClock.GetSlotTime(targetSlot).Add(unpreparedGrabOffset(attestationDue(a.beaconChainCfg, stateVersion)))
+ boundedCtx, cancel := context.WithDeadline(ctx, collectAt)
+ defer cancel()
+
+ for time.Now().Before(collectAt) {
+ idBytes, err := a.engine.ForkChoiceUpdate(boundedCtx, finalized, safe, head, attrs, stateVersion)
+ switch {
+ case err == nil:
+ return idBytes, nil
+ case ctx.Err() != nil:
+ return nil, ctx.Err()
+ case !errors.Is(err, execution_client.ErrForkChoiceBusy) && !errors.Is(err, context.DeadlineExceeded):
+ return nil, err
+ }
+ select {
+ case <-ctx.Done():
+ return nil, ctx.Err()
+ case <-time.After(forkChoiceBusyRetryDelay):
+ }
+ }
+ return a.engine.ForkChoiceUpdate(ctx, finalized, safe, head, attrs, stateVersion)
+}
+
+// payloadAttributes builds the attributes preparation and production must agree on. Withdrawals and
+// the parent beacon block root are set for every fork because the execution layer decides what to do
+// with them from the payload timestamp; gating them on the consensus fork instead puts the two sides
+// on different oracles, which disagree on any chain where the forks are not aligned.
+func payloadAttributes(
+ timestamp hexutil.Uint64,
+ prevRandao common.Hash,
+ feeRecipient common.Address,
+ withdrawals []*types.Withdrawal,
+ parentRoot *common.Hash,
+) *engine_types.PayloadAttributes {
+ return &engine_types.PayloadAttributes{
+ Timestamp: timestamp,
+ PrevRandao: prevRandao,
+ SuggestedFeeRecipient: feeRecipient,
+ Withdrawals: withdrawals,
+ ParentBeaconBlockRoot: parentRoot,
+ }
+}
+
+// gloasWithdrawals reads the withdrawals for a Gloas proposal from the state copy carrying the
+// parent payload when the head is FULL, and from the cached expectation when it is EMPTY.
+func gloasWithdrawals(baseState, withParentPayload *state.CachingBeaconState, epoch uint64) ([]*types.Withdrawal, error) {
+ if withParentPayload != nil {
+ clWithdrawals, err := state.GetExpectedWithdrawals(withParentPayload, epoch)
+ if err != nil {
+ return nil, err
+ }
+ return cltypes.ConvertConsensusWithdrawalsToExecutionWithdrawals(clWithdrawals.Withdrawals), nil
+ }
+ cached := baseState.GetPayloadExpectedWithdrawals()
+ if cached == nil {
+ return nil, nil
+ }
+ consensusWithdrawals := make([]*cltypes.Withdrawal, cached.Len())
+ for i := range consensusWithdrawals {
+ consensusWithdrawals[i] = cached.Get(i)
+ }
+ return cltypes.ConvertConsensusWithdrawalsToExecutionWithdrawals(consensusWithdrawals), nil
+}
+
func shouldRetryGetPayload(now, deadline time.Time) bool {
return now.Before(deadline)
}
@@ -252,12 +351,30 @@ func pollAssembledPayload(
defer deadlineTimer.Stop()
retryTicker := time.NewTicker(retryTime)
defer retryTicker.Stop()
+ // A slot that fails tends to fail on every poll of the window, so report the first and then a
+ // count rather than hundreds of copies of one line. Contention that clears is not worth an
+ // alert, so the summary is only for a window that never produced anything.
+ var (
+ failures int
+ lastErr error
+ collected bool
+ )
+ defer func() {
+ if failures > 1 && !collected {
+ log.Error("BlockProduction: payload polling kept failing", "attempts", failures, "err", lastErr)
+ }
+ }()
for {
// Grab at least once, even past the deadline, so a late produce request still gets a payload.
payload, bundles, requestsBundle, blockValue, err := get()
if err != nil {
- log.Error("BlockProduction: Failed to get payload", "err", err)
+ failures++
+ lastErr = err
+ if failures == 1 {
+ log.Error("BlockProduction: Failed to get payload", "err", err)
+ }
} else if payload != nil {
+ collected = true
return payload, bundles, requestsBundle, blockValue, true
}
select {
@@ -473,10 +590,6 @@ func (a *ApiHandler) GetEthV3ValidatorBlock(
}); err != nil {
return nil, err
}
-
- if err != nil {
- return nil, err
- }
if baseState == nil {
return nil, beaconhttp.NewEndpointError(
http.StatusNotFound,
@@ -863,6 +976,9 @@ func (a *ApiHandler) produceBeaconBody(
baseBlockSlot,
)
}
+ a.proposalsInFlight.Add(1)
+ defer a.proposalsInFlight.Add(-1)
+
var wg sync.WaitGroup
stateVersion := a.beaconChainCfg.GetCurrentStateVersion(
targetSlot / a.beaconChainCfg.SlotsPerEpoch,
@@ -924,14 +1040,7 @@ func (a *ApiHandler) produceBeaconBody(
head = baseState.GetLatestBlockHash()
}
}
- finalizedHash := a.forkchoiceStore.GetFinalizedExecutionHash(baseState.FinalizedCheckpoint().Root)
- if finalizedHash == (common.Hash{}) {
- finalizedHash = head
- }
- safeHash := a.forkchoiceStore.GetFinalizedExecutionHash(baseState.CurrentJustifiedCheckpoint().Root)
- if safeHash == (common.Hash{}) {
- safeHash = head
- }
+ safeHash, finalizedHash := a.executionCheckpointHashes(baseState, head)
proposerIndex, err := baseState.GetBeaconProposerIndexForSlot(targetSlot)
if err != nil {
return nil, 0, err
@@ -953,8 +1062,8 @@ func (a *ApiHandler) produceBeaconBody(
}
}
}
- currEpoch := a.ethClock.GetCurrentEpoch()
- random := baseState.GetRandaoMixes(currEpoch)
+ targetEpoch := targetSlot / a.beaconChainCfg.SlotsPerEpoch
+ random := baseState.GetRandaoMixes(targetEpoch)
var executionPayload *cltypes.Eth1Block
var executionValue uint64
@@ -975,85 +1084,46 @@ func (a *ApiHandler) produceBeaconBody(
log.Info("BlockProduction: ForkChoiceUpdate&GetPayload took", "duration", time.Since(start))
}()
retryTime := 10 * time.Millisecond
- feeRecipient, _ := a.validatorParams.GetFeeRecipient(proposerIndex)
- var withdrawals []*types.Withdrawal
- switch {
- case gloasWithdrawalsState != nil:
- // GLOAS FULL: compute withdrawals from the state copy with parent payload applied
- clWithdrawals, err := state.GetExpectedWithdrawals(
- gloasWithdrawalsState,
- targetSlot/a.beaconChainCfg.SlotsPerEpoch,
- )
+ feeRecipient, registered := a.validatorParams.GetFeeRecipient(proposerIndex)
+ if !registered {
+ // Preparation treats an unregistered proposer as someone else's slot; production
+ // still builds, so make the zero-address fallback visible rather than silent.
+ log.Warn("BlockProduction: no fee recipient registered for proposer, using zero address",
+ "proposer", proposerIndex)
+ }
+ // One branch per fork: the pre-Gloas inputs are the same ones preparation builds, and the
+ // prepared-id match depends on production not deriving its own variant of them.
+ var (
+ fcuHead, fcuSafeHash, fcuFinalizedHash common.Hash
+ attrs *engine_types.PayloadAttributes
+ )
+ if stateVersion.Before(clparams.GloasVersion) {
+ var err error
+ fcuHead, fcuSafeHash, fcuFinalizedHash, attrs, err = a.preGloasForkChoiceInputs(baseState, baseBlockRoot, targetSlot, feeRecipient)
if err != nil {
- log.Error("BlockProduction: GetExpectedWithdrawals (FULL) failed", "err", err)
+ log.Error("BlockProduction: build forkchoice inputs failed", "err", err)
return
}
- withdrawals = make([]*types.Withdrawal, 0, len(clWithdrawals.Withdrawals))
- for _, w := range clWithdrawals.Withdrawals {
- withdrawals = append(withdrawals, &types.Withdrawal{
- Index: w.Index,
- Amount: w.Amount,
- Validator: w.Validator,
- Address: w.Address,
- })
- }
- case stateVersion >= clparams.GloasVersion && gloasWithdrawalsState == nil:
- // GLOAS EMPTY: use cached payload_expected_withdrawals from state
- cachedWithdrawals := baseState.GetPayloadExpectedWithdrawals()
- if cachedWithdrawals != nil {
- withdrawals = make([]*types.Withdrawal, 0, cachedWithdrawals.Len())
- for i := 0; i < cachedWithdrawals.Len(); i++ {
- w := cachedWithdrawals.Get(i)
- withdrawals = append(withdrawals, &types.Withdrawal{
- Index: w.Index,
- Amount: w.Amount,
- Validator: w.Validator,
- Address: w.Address,
- })
- }
- }
- default:
- // Pre-GLOAS: compute withdrawals normally
- clWithdrawals, err := state.GetExpectedWithdrawals(
- baseState,
- targetSlot/a.beaconChainCfg.SlotsPerEpoch,
- )
+ } else {
+ fcuHead, fcuSafeHash, fcuFinalizedHash = head, safeHash, finalizedHash
+ withdrawals, err := gloasWithdrawals(baseState, gloasWithdrawalsState, targetSlot/a.beaconChainCfg.SlotsPerEpoch)
if err != nil {
log.Error("BlockProduction: GetExpectedWithdrawals failed", "err", err)
return
}
- withdrawals = make([]*types.Withdrawal, 0, len(clWithdrawals.Withdrawals))
- for _, w := range clWithdrawals.Withdrawals {
- withdrawals = append(withdrawals, &types.Withdrawal{
- Index: w.Index,
- Amount: w.Amount,
- Validator: w.Validator,
- Address: w.Address,
- })
- }
- }
-
- attrs := &engine_types.PayloadAttributes{
- Timestamp: hexutil.Uint64(state.ComputeTimestampAtSlot(baseState, targetSlot)),
- PrevRandao: random,
- SuggestedFeeRecipient: feeRecipient,
- Withdrawals: withdrawals,
- ParentBeaconBlockRoot: (*common.Hash)(&blockRoot),
- }
- if stateVersion.AfterOrEqual(clparams.GloasVersion) {
- sn := hexutil.Uint64(targetSlot)
- attrs.SlotNumber = &sn
+ attrs = payloadAttributes(
+ hexutil.Uint64(state.ComputeTimestampAtSlot(baseState, targetSlot)),
+ random,
+ feeRecipient,
+ withdrawals,
+ (*common.Hash)(&blockRoot),
+ )
+ slotNumber := hexutil.Uint64(targetSlot)
+ attrs.SlotNumber = &slotNumber
attrs.TargetGasLimit = targetGasLimit
}
builderStartedAt := time.Now()
- idBytes, err := a.engine.ForkChoiceUpdate(
- ctx,
- finalizedHash,
- safeHash,
- head,
- attrs,
- stateVersion,
- )
+ idBytes, err := a.forkChoiceUpdateForProposal(ctx, targetSlot, fcuFinalizedHash, fcuSafeHash, fcuHead, attrs, stateVersion)
if err != nil {
log.Error("BlockProduction: Failed to get payload id", "err", err)
return
@@ -1063,7 +1133,16 @@ func (a *ApiHandler) produceBeaconBody(
return
}
slotStart := a.ethClock.GetSlotTime(targetSlot)
- buildWindow := computeBlockBuilderWindow(builderStartedAt, slotStart, a.beaconChainCfg, stateVersion)
+ prepared := canUsePreparedPayload(
+ &a.preparedPayload,
+ a.engine.SupportInsertion(),
+ targetSlot,
+ idBytes,
+ time.Now(),
+ preparedPayloadMinimumAge(a.beaconChainCfg, stateVersion),
+ )
+ log.Info("BlockProduction: payload preparation", "slot", targetSlot, "prepared", prepared)
+ buildWindow := computeBlockBuilderWindow(builderStartedAt, slotStart, a.beaconChainCfg, stateVersion, prepared)
payload, bundles, requestsBundle, blockValue, ok := pollAssembledPayload(ctx, buildWindow, retryTime, func() (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error) {
return a.engine.GetAssembledBlock(ctx, idBytes, stateVersion)
})
@@ -2592,15 +2671,12 @@ func (a *ApiHandler) cacheExecutionBody(payload *cltypes.Eth1Block) {
}
var ws []*types.Withdrawal
if payload.Withdrawals != nil {
- payload.Withdrawals.Range(func(idx int, w *cltypes.Withdrawal, total int) bool {
- ws = append(ws, &types.Withdrawal{
- Index: w.Index,
- Validator: w.Validator,
- Address: w.Address,
- Amount: w.Amount,
- })
+ consensusWithdrawals := make([]*cltypes.Withdrawal, 0, payload.Withdrawals.Len())
+ payload.Withdrawals.Range(func(_ int, w *cltypes.Withdrawal, _ int) bool {
+ consensusWithdrawals = append(consensusWithdrawals, w)
return true
})
+ ws = cltypes.ConvertConsensusWithdrawalsToExecutionWithdrawals(consensusWithdrawals)
}
a.blockReader.CacheBlockBody(payload.BlockNumber, rawTxs, ws)
}
diff --git a/cl/beacon/handler/block_production_test.go b/cl/beacon/handler/block_production_test.go
index 6f6951bc734..07f09113f0b 100644
--- a/cl/beacon/handler/block_production_test.go
+++ b/cl/beacon/handler/block_production_test.go
@@ -27,16 +27,23 @@ import (
"testing"
"time"
+ "github.com/go-chi/chi/v5"
"github.com/holiman/uint256"
"github.com/stretchr/testify/require"
"go.uber.org/mock/gomock"
+ "github.com/erigontech/erigon/cl/beacon/beacon_router_configuration"
"github.com/erigontech/erigon/cl/beacon/beaconhttp"
builder_mock "github.com/erigontech/erigon/cl/beacon/builder/mock_services"
+ "github.com/erigontech/erigon/cl/beacon/synced_data"
+ sync_mock_services "github.com/erigontech/erigon/cl/beacon/synced_data/mock_services"
"github.com/erigontech/erigon/cl/clparams"
"github.com/erigontech/erigon/cl/cltypes"
"github.com/erigontech/erigon/cl/cltypes/solid"
+ "github.com/erigontech/erigon/cl/phase1/core/state"
"github.com/erigontech/erigon/cl/phase1/execution_client"
+ "github.com/erigontech/erigon/cl/transition"
+ "github.com/erigontech/erigon/cl/utils/eth_clock"
"github.com/erigontech/erigon/common"
"github.com/erigontech/erigon/common/hexutil"
"github.com/erigontech/erigon/common/log/v3"
@@ -59,7 +66,7 @@ func TestBlockBuilderWindowPreGloas(t *testing.T) {
slotStart := time.Unix(100, 0)
now := slotStart
- window := computeBlockBuilderWindow(now, slotStart, cfg, clparams.ElectraVersion)
+ window := computeBlockBuilderWindow(now, slotStart, cfg, clparams.ElectraVersion, false)
// Attestation deadline is 4s; polling stops a quarter of it (1s) earlier, at 3s.
require.Equal(t, slotStart.Add(3*time.Second).Add(-minPayloadPollingWindow), window.firstGetAt)
@@ -74,7 +81,7 @@ func TestBlockBuilderWindowGloas(t *testing.T) {
slotStart := time.Unix(100, 0)
now := slotStart
- window := computeBlockBuilderWindow(now, slotStart, cfg, clparams.GloasVersion)
+ window := computeBlockBuilderWindow(now, slotStart, cfg, clparams.GloasVersion, false)
// Attestation deadline is 3s; polling stops a quarter of it (750ms) earlier, at 2.25s.
require.Equal(t, slotStart.Add(2250*time.Millisecond).Add(-minPayloadPollingWindow), window.firstGetAt)
@@ -327,7 +334,7 @@ func TestBlockBuilderWindowLateStartKeepsPublicationMargin(t *testing.T) {
slotStart := time.Unix(100, 0)
now := slotStart.Add(2950 * time.Millisecond)
- window := computeBlockBuilderWindow(now, slotStart, cfg, clparams.ElectraVersion)
+ window := computeBlockBuilderWindow(now, slotStart, cfg, clparams.ElectraVersion, false)
// A late request clamps the first poll up to now but still stops at 3s, preserving the margin.
require.Equal(t, now, window.firstGetAt)
@@ -342,7 +349,7 @@ func TestBlockBuilderWindowLateRequestGrabsImmediately(t *testing.T) {
slotStart := time.Unix(100, 0)
now := slotStart.Add(5 * time.Second)
- window := computeBlockBuilderWindow(now, slotStart, cfg, clparams.GloasVersion)
+ window := computeBlockBuilderWindow(now, slotStart, cfg, clparams.GloasVersion, false)
require.Equal(t, now, window.firstGetAt)
require.Equal(t, now, window.pollUntil)
@@ -365,7 +372,7 @@ func TestBlockBuilderWindowReservesPublicationMargin(t *testing.T) {
{"gloas", clparams.GloasVersion, 3 * time.Second, 2250 * time.Millisecond},
} {
t.Run(tc.name, func(t *testing.T) {
- window := computeBlockBuilderWindow(slotStart, slotStart, cfg, tc.version)
+ window := computeBlockBuilderWindow(slotStart, slotStart, cfg, tc.version, false)
require.Equal(t, slotStart.Add(tc.wantPollUntil), window.pollUntil)
require.True(t, window.pollUntil.Before(slotStart.Add(tc.deadline)),
"polling must stop before the attestation deadline to leave publication margin")
@@ -732,3 +739,586 @@ func TestCaplinBlockProductionGlamsterdamSlotNumber(t *testing.T) {
require.Equal(t, hexutil.Uint64(targetSlot), *spy.lastAttributes.SlotNumber,
"SlotNumber should equal the target slot")
}
+
+func TestBlockBuilderWindowTakesPreparedPayloadEarly(t *testing.T) {
+ cfg := &clparams.BeaconChainConfig{
+ SecondsPerSlot: 12,
+ IntervalsPerSlot: 3,
+ }
+ slotStart := time.Unix(100, 0)
+
+ // A primed payload is taken one quarter of the attestation deadline into the slot.
+ prepared := computeBlockBuilderWindow(slotStart, slotStart, cfg, clparams.ElectraVersion, true)
+ require.Equal(t, slotStart.Add(time.Second).Add(-minPayloadPollingWindow), prepared.firstGetAt)
+ require.Equal(t, slotStart.Add(3*time.Second), prepared.pollUntil)
+
+ // Without a primed builder nothing changes: the execution layer still needs most of the slot.
+ unprepared := computeBlockBuilderWindow(slotStart, slotStart, cfg, clparams.ElectraVersion, false)
+ require.Equal(t, slotStart.Add(3*time.Second).Add(-minPayloadPollingWindow), unprepared.firstGetAt)
+ require.Equal(t, slotStart.Add(3*time.Second), unprepared.pollUntil)
+
+ require.True(t, prepared.firstGetAt.Before(unprepared.firstGetAt))
+
+ late := computeBlockBuilderWindow(slotStart.Add(2*time.Second), slotStart, cfg, clparams.ElectraVersion, true)
+ require.Equal(t, slotStart.Add(2*time.Second), late.firstGetAt)
+ require.Equal(t, slotStart.Add(3*time.Second), late.pollUntil)
+}
+
+func TestPreparedPayloadMatchesOnlyTheSamePrime(t *testing.T) {
+ var p preparedPayload
+ id := []byte{1, 2, 3, 4, 5, 6, 7, 8}
+ now := time.Unix(100, 0)
+
+ require.False(t, p.matches(10, id, now, 0), "nothing primed yet")
+
+ p.set(10, id, now)
+ require.True(t, p.matches(10, id, now, 0))
+
+ // A different payload id means the execution layer started a fresh build — a reorg, a late
+ // block, or a changed fee recipient — so the warm builder is gone.
+ require.False(t, p.matches(10, []byte{9, 9, 9, 9, 9, 9, 9, 9}, now, 0))
+ require.False(t, p.matches(11, id, now, 0), "primed for another slot")
+ require.False(t, p.matches(10, nil, now, 0), "no id from the execution layer")
+}
+
+func TestPreparedPayloadRequiresMinimumWarmup(t *testing.T) {
+ var p preparedPayload
+ id := []byte{1, 2, 3, 4, 5, 6, 7, 8}
+ now := time.Unix(100, 0)
+ minAge := 2 * time.Second
+
+ p.set(10, id, now.Add(-minAge+time.Nanosecond))
+ require.False(t, p.matches(10, id, now, minAge))
+
+ p.set(10, id, now.Add(-minAge))
+ require.True(t, p.matches(10, id, now, minAge))
+}
+
+func TestPreparedPayloadMinimumWarmupPreservesBuildTime(t *testing.T) {
+ cfg := &clparams.BeaconChainConfig{
+ SecondsPerSlot: 12,
+ IntervalsPerSlot: 3,
+ }
+
+ require.Equal(t, 2*time.Second, preparedPayloadMinimumAge(cfg, clparams.ElectraVersion))
+}
+
+func TestPreparedPayloadRequiresBuilderContinuity(t *testing.T) {
+ var p preparedPayload
+ id := []byte{1, 2, 3, 4, 5, 6, 7, 8}
+ now := time.Unix(100, 0)
+ p.set(10, id, now.Add(-3*time.Second))
+
+ require.True(t, canUsePreparedPayload(&p, true, 10, id, now, 2*time.Second))
+ require.False(t, canUsePreparedPayload(&p, false, 10, id, now, 2*time.Second))
+}
+
+func TestStartPayloadPreparationSkipsRemoteEngine(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ engine := execution_client.NewMockExecutionEngine(ctrl)
+ engine.EXPECT().SupportInsertion().Return(false)
+ handler := &ApiHandler{
+ engine: engine,
+ routerCfg: &beacon_router_configuration.RouterConfiguration{Validator: true},
+ beaconChainCfg: &clparams.BeaconChainConfig{SecondsPerSlot: 12},
+ }
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+
+ handler.StartPayloadPreparation(ctx)
+}
+
+func TestStartPayloadPreparationSkipsWithoutValidatorAPI(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ engine := execution_client.NewMockExecutionEngine(ctrl)
+ engine.EXPECT().SupportInsertion().Times(0)
+ handler := &ApiHandler{
+ engine: engine,
+ routerCfg: &beacon_router_configuration.RouterConfiguration{Validator: false},
+ beaconChainCfg: &clparams.BeaconChainConfig{SecondsPerSlot: 12},
+ }
+
+ handler.StartPayloadPreparation(t.Context())
+}
+
+func TestStartPayloadPreparationSkipsNilEngine(t *testing.T) {
+ handler := &ApiHandler{}
+ require.NotPanics(t, func() {
+ handler.StartPayloadPreparation(t.Context())
+ })
+}
+
+func TestStartPayloadPreparationStartsLocalEngine(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ engine := execution_client.NewMockExecutionEngine(ctrl)
+ engine.EXPECT().SupportInsertion().Return(true)
+ handler := &ApiHandler{
+ engine: engine,
+ routerCfg: &beacon_router_configuration.RouterConfiguration{Validator: true},
+ beaconChainCfg: &clparams.BeaconChainConfig{SecondsPerSlot: 12},
+ }
+ ctx, cancel := context.WithCancel(t.Context())
+ cancel()
+
+ handler.StartPayloadPreparation(ctx)
+}
+
+func TestPreparePayloadLoopRunsImmediatelyWithSlotDeadline(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ _, _, _, _, postState, handler, _, _, _, validatorParams := setupTestingHandler(t, clparams.ElectraVersion, log.Root(), true)
+ targetSlot := postState.Slot() + 1
+ proposerIndex, err := postState.GetBeaconProposerIndexForSlot(targetSlot)
+ require.NoError(t, err)
+ validatorParams.SetFeeRecipient(proposerIndex, common.Address{0x11})
+
+ slotStart := time.Now().Add(6 * time.Second)
+ clock := eth_clock.NewMockEthereumClock(ctrl)
+ clock.EXPECT().GetCurrentSlot().Return(targetSlot - 1)
+ clock.EXPECT().GetSlotTime(targetSlot).Return(slotStart).AnyTimes()
+ handler.ethClock = clock
+
+ ctx, cancel := context.WithCancel(t.Context())
+ engine := execution_client.NewMockExecutionEngine(ctrl)
+ engine.EXPECT().ForkChoiceUpdate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
+ DoAndReturn(func(callCtx context.Context, _, _, _ common.Hash, _ *engine_types.PayloadAttributes, _ clparams.StateVersion) ([]byte, error) {
+ deadline, ok := callCtx.Deadline()
+ require.True(t, ok)
+ require.Equal(t, slotStart, deadline)
+ cancel()
+ return []byte{1, 2, 3, 4, 5, 6, 7, 8}, nil
+ })
+ handler.engine = engine
+
+ handler.preparePayloadLoop(ctx)
+}
+
+func TestPreparePayloadLoopSkipsSlotsTooFarAhead(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ _, _, _, _, postState, handler, _, _, _, validatorParams := setupTestingHandler(t, clparams.ElectraVersion, log.Root(), true)
+ targetSlot := postState.Slot() + 1
+ proposerIndex, err := postState.GetBeaconProposerIndexForSlot(targetSlot)
+ require.NoError(t, err)
+ validatorParams.SetFeeRecipient(proposerIndex, common.Address{0x11})
+
+ // Before genesis the current slot clamps to zero, so the next slot can be hours out.
+ clock := eth_clock.NewMockEthereumClock(ctrl)
+ clock.EXPECT().GetCurrentSlot().Return(targetSlot - 1).AnyTimes()
+ clock.EXPECT().GetSlotTime(targetSlot).Return(time.Now().Add(time.Hour)).AnyTimes()
+ handler.ethClock = clock
+
+ engine := execution_client.NewMockExecutionEngine(ctrl)
+ engine.EXPECT().SupportInsertion().Return(true)
+ engine.EXPECT().ForkChoiceUpdate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Times(0)
+ handler.engine = engine
+
+ ctx, cancel := context.WithTimeout(t.Context(), 250*time.Millisecond)
+ defer cancel()
+ handler.StartPayloadPreparation(ctx)
+ <-ctx.Done()
+}
+
+func TestPreparePayloadLoopStandsOffWhileProducing(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ _, _, _, _, postState, handler, _, _, _, validatorParams := setupTestingHandler(t, clparams.ElectraVersion, log.Root(), true)
+ targetSlot := postState.Slot() + 1
+ proposerIndex, err := postState.GetBeaconProposerIndexForSlot(targetSlot)
+ require.NoError(t, err)
+ validatorParams.SetFeeRecipient(proposerIndex, common.Address{0x11})
+
+ clock := eth_clock.NewMockEthereumClock(ctrl)
+ clock.EXPECT().GetCurrentSlot().Return(targetSlot - 1).AnyTimes()
+ clock.EXPECT().GetSlotTime(gomock.Any()).Times(0)
+ handler.ethClock = clock
+
+ // Priming would contend with the block being produced for the execution layer's single slot.
+ handler.proposalsInFlight.Add(1)
+ engine := execution_client.NewMockExecutionEngine(ctrl)
+ engine.EXPECT().ForkChoiceUpdate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Times(0)
+ handler.engine = engine
+
+ ctx, cancel := context.WithTimeout(t.Context(), 250*time.Millisecond)
+ defer cancel()
+ handler.preparePayloadLoop(ctx)
+}
+
+func TestPreparePayloadLoopSkipsSlotsAboutToStart(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ _, _, _, _, postState, handler, _, _, _, validatorParams := setupTestingHandler(t, clparams.ElectraVersion, log.Root(), true)
+ targetSlot := postState.Slot() + 1
+ proposerIndex, err := postState.GetBeaconProposerIndexForSlot(targetSlot)
+ require.NoError(t, err)
+ validatorParams.SetFeeRecipient(proposerIndex, common.Address{0x11})
+
+ // Too close to the slot for the prime to ever reach the age production demands of it, so the
+ // state copy and the forkchoice update would both be spent for nothing.
+ clock := eth_clock.NewMockEthereumClock(ctrl)
+ clock.EXPECT().GetCurrentSlot().Return(targetSlot - 1).AnyTimes()
+ clock.EXPECT().GetSlotTime(targetSlot).Return(time.Now().Add(200 * time.Millisecond)).AnyTimes()
+ handler.ethClock = clock
+
+ engine := execution_client.NewMockExecutionEngine(ctrl)
+ engine.EXPECT().ForkChoiceUpdate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Times(0)
+ handler.engine = engine
+
+ ctx, cancel := context.WithTimeout(t.Context(), 250*time.Millisecond)
+ defer cancel()
+ handler.preparePayloadLoop(ctx)
+}
+
+func TestForkChoiceUpdateForProposalWaitsOutContention(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ clock := eth_clock.NewMockEthereumClock(ctrl)
+ clock.EXPECT().GetSlotTime(uint64(10)).Return(time.Now())
+
+ calls := 0
+ engine := execution_client.NewMockExecutionEngine(ctrl)
+ engine.EXPECT().ForkChoiceUpdate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
+ DoAndReturn(func(context.Context, common.Hash, common.Hash, common.Hash, *engine_types.PayloadAttributes, clparams.StateVersion) ([]byte, error) {
+ calls++
+ if calls < 3 {
+ return nil, execution_client.ErrForkChoiceBusy
+ }
+ return []byte{1, 2, 3, 4, 5, 6, 7, 8}, nil
+ }).Times(3)
+
+ a := &ApiHandler{engine: engine, ethClock: clock, beaconChainCfg: &clparams.MainnetBeaconConfig}
+ id, err := a.forkChoiceUpdateForProposal(t.Context(), 10, common.Hash{}, common.Hash{}, common.Hash{}, &engine_types.PayloadAttributes{}, clparams.ElectraVersion)
+
+ require.NoError(t, err)
+ require.Len(t, id, 8)
+}
+
+func TestForkChoiceUpdateForProposalStopsOnceTheWindowClosed(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ clock := eth_clock.NewMockEthereumClock(ctrl)
+ clock.EXPECT().GetSlotTime(uint64(10)).Return(time.Now().Add(-time.Minute))
+
+ engine := execution_client.NewMockExecutionEngine(ctrl)
+ engine.EXPECT().ForkChoiceUpdate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
+ Return(nil, execution_client.ErrForkChoiceBusy).Times(1)
+
+ // Past the point the payload has to be collected, another attempt cannot help this slot.
+ a := &ApiHandler{engine: engine, ethClock: clock, beaconChainCfg: &clparams.MainnetBeaconConfig}
+ _, err := a.forkChoiceUpdateForProposal(t.Context(), 10, common.Hash{}, common.Hash{}, common.Hash{}, &engine_types.PayloadAttributes{}, clparams.ElectraVersion)
+
+ require.ErrorIs(t, err, execution_client.ErrForkChoiceBusy)
+}
+
+func TestPreparePayloadForSendsCompleteForkChoiceUpdate(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ _, _, _, _, postState, handler, _, _, fcu, validatorParams := setupTestingHandler(t, clparams.CapellaVersion, log.Root(), true)
+ config := *handler.beaconChainCfg
+ targetEpoch := postState.Slot()/config.SlotsPerEpoch + 1
+ targetSlot := targetEpoch * config.SlotsPerEpoch
+ config.DenebForkEpoch = targetEpoch
+ config.InitializeForkSchedule()
+
+ headState := state.New(&config)
+ require.NoError(t, postState.CopyInto(headState))
+ headState.SetFinalizedCheckpoint(solid.Checkpoint{Epoch: targetEpoch - 2, Root: common.Hash{0x31}})
+ headState.SetCurrentJustifiedCheckpoint(solid.Checkpoint{Epoch: targetEpoch - 1, Root: common.Hash{0x32}})
+ currentEpoch := headState.Slot() / config.SlotsPerEpoch
+ headState.SetRandaoMixAt(int(currentEpoch%config.EpochsPerHistoricalVector), common.Hash{0x51})
+ headState.SetRandaoMixAt(int(targetEpoch%config.EpochsPerHistoricalVector), common.Hash{0x52})
+ baseBlockRoot := common.Hash{0x41}
+ syncedData := synced_data.NewSyncedDataManager(&config, true)
+ require.NoError(t, syncedData.OnHeadStateWithBlockRoot(headState, baseBlockRoot))
+ syncedData.OnSelectedHead(baseBlockRoot, headState.Slot())
+ handler.beaconChainCfg = &config
+ handler.syncedData = syncedData
+
+ proposerIndex, err := headState.GetBeaconProposerIndexForSlot(targetSlot)
+ require.NoError(t, err)
+
+ feeRecipient := common.Address{0x11}
+ validatorParams.SetFeeRecipient(proposerIndex, feeRecipient)
+
+ advancedState, err := headState.Copy()
+ require.NoError(t, err)
+ require.NoError(t, transition.DefaultMachine.ProcessSlots(advancedState, targetSlot))
+ require.Equal(t, clparams.CapellaVersion, headState.Version())
+ require.Equal(t, clparams.DenebVersion, advancedState.Version())
+ require.NotEqual(t, headState.GetRandaoMixes(targetEpoch), advancedState.GetRandaoMixes(targetEpoch))
+ finalizedRoot := advancedState.FinalizedCheckpoint().Root
+ justifiedRoot := advancedState.CurrentJustifiedCheckpoint().Root
+ require.NotEqual(t, finalizedRoot, justifiedRoot)
+ expectedFinalized := common.Hash{0x21}
+ expectedSafe := common.Hash{0x22}
+ fcu.Eth1Hashes[finalizedRoot] = expectedFinalized
+ fcu.Eth1Hashes[justifiedRoot] = expectedSafe
+ require.NotEqual(t, expectedFinalized, expectedSafe)
+
+ version := handler.beaconChainCfg.GetCurrentStateVersion(targetSlot / handler.beaconChainCfg.SlotsPerEpoch)
+ require.Equal(t, clparams.DenebVersion, version)
+ expectedWithdrawals, err := state.GetExpectedWithdrawals(advancedState, targetSlot/handler.beaconChainCfg.SlotsPerEpoch)
+ require.NoError(t, err)
+
+ payloadID := []byte{1, 2, 3, 4, 5, 6, 7, 8}
+ engine := execution_client.NewMockExecutionEngine(ctrl)
+ engine.EXPECT().ForkChoiceUpdate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
+ DoAndReturn(func(_ context.Context, finalized, safe, head common.Hash, attrs *engine_types.PayloadAttributes, gotVersion clparams.StateVersion) ([]byte, error) {
+ require.Equal(t, expectedFinalized, finalized)
+ require.Equal(t, expectedSafe, safe)
+ require.Equal(t, advancedState.LatestExecutionPayloadHeader().BlockHash, head)
+ require.Equal(t, version, gotVersion)
+ require.Equal(t, hexutil.Uint64(state.ComputeTimestampAtSlot(advancedState, targetSlot)), attrs.Timestamp)
+ require.Equal(t, common.Hash(advancedState.GetRandaoMixes(targetSlot/handler.beaconChainCfg.SlotsPerEpoch)), attrs.PrevRandao)
+ require.Equal(t, feeRecipient, attrs.SuggestedFeeRecipient)
+ require.Equal(t, &baseBlockRoot, attrs.ParentBeaconBlockRoot)
+ require.Nil(t, attrs.SlotNumber)
+ require.Nil(t, attrs.TargetGasLimit)
+ require.NotNil(t, attrs.Withdrawals)
+ require.Len(t, attrs.Withdrawals, len(expectedWithdrawals.Withdrawals))
+ for i, withdrawal := range expectedWithdrawals.Withdrawals {
+ require.Equal(t, withdrawal.Index, attrs.Withdrawals[i].Index)
+ require.Equal(t, withdrawal.Amount, attrs.Withdrawals[i].Amount)
+ require.Equal(t, withdrawal.Validator, attrs.Withdrawals[i].Validator)
+ require.Equal(t, withdrawal.Address, attrs.Withdrawals[i].Address)
+ }
+ return payloadID, nil
+ })
+ handler.engine = engine
+
+ clock := eth_clock.NewMockEthereumClock(ctrl)
+ clock.EXPECT().GetSlotTime(gomock.Any()).Return(time.Now().Add(6 * time.Second)).AnyTimes()
+ handler.ethClock = clock
+
+ primedHead, err := handler.preparePayloadFor(t.Context(), targetSlot)
+ require.NoError(t, err)
+ require.Equal(t, baseBlockRoot, primedHead)
+ require.True(t, handler.preparedPayload.matches(targetSlot, payloadID, time.Now(), 0))
+}
+
+func TestPreparePayloadForRejectsChangedHeadBeforeForkChoiceUpdate(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ _, _, _, _, postState, handler, _, syncedData, _, validatorParams := setupTestingHandler(t, clparams.ElectraVersion, log.Root(), false)
+ targetSlot := postState.Slot() + 1
+ proposerIndex, err := postState.GetBeaconProposerIndexForSlot(targetSlot)
+ require.NoError(t, err)
+ validatorParams.SetFeeRecipient(proposerIndex, common.Address{0x11})
+
+ baseBlockRoot := common.Hash{0x41}
+ changedBlockRoot := common.Hash{0x42}
+ syncedDataMock := syncedData.(*sync_mock_services.MockSyncedData)
+ gomock.InOrder(
+ syncedDataMock.EXPECT().ViewHeadStateWithIdentity(gomock.Any()).
+ DoAndReturn(func(view synced_data.ViewHeadStateWithIdentityFn) error {
+ return view(postState, baseBlockRoot, postState.Slot())
+ }),
+ syncedDataMock.EXPECT().SelectedHead().Return(changedBlockRoot, postState.Slot(), true),
+ )
+ engine := execution_client.NewMockExecutionEngine(ctrl)
+ engine.EXPECT().ForkChoiceUpdate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Times(0)
+ handler.engine = engine
+
+ clock := eth_clock.NewMockEthereumClock(ctrl)
+ clock.EXPECT().GetSlotTime(gomock.Any()).Return(time.Now().Add(6 * time.Second)).AnyTimes()
+ handler.ethClock = clock
+
+ _, err = handler.preparePayloadFor(t.Context(), targetSlot)
+ require.ErrorIs(t, err, errPreparationHeadChanged)
+ require.False(t, handler.preparedPayload.matches(targetSlot, []byte{1}, time.Now(), 0))
+}
+
+func TestPreparePayloadForUsesPostEpochProposer(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ _, _, _, _, postState, handler, _, syncedData, _, validatorParams := setupTestingHandler(t, clparams.ElectraVersion, log.Root(), false)
+ config := *handler.beaconChainCfg
+ targetEpoch := postState.Slot()/config.SlotsPerEpoch + 1
+ targetSlot := targetEpoch * config.SlotsPerEpoch
+ config.FuluForkEpoch = targetEpoch + 10
+ config.GloasForkEpoch = targetEpoch + 11
+ config.InitializeForkSchedule()
+ handler.beaconChainCfg = &config
+
+ headState := state.New(&config)
+ require.NoError(t, postState.CopyInto(headState))
+ headState.SetSlot(targetSlot - 1)
+ for i := 0; i < headState.ValidatorLength(); i += 2 {
+ headState.SetEffectiveBalanceForValidatorAtIndex(i, 0)
+ require.NoError(t, headState.SetValidatorBalance(i, config.MaxEffectiveBalanceElectra))
+ }
+
+ mixPosition := (targetEpoch + config.EpochsPerHistoricalVector - config.MinSeedLookahead - 1) % config.EpochsPerHistoricalVector
+ var oldProposer, newProposer uint64
+ found := false
+ for nonce := byte(0); ; nonce++ {
+ headState.SetRandaoMixAt(int(mixPosition), common.Hash{nonce})
+ var err error
+ oldProposer, err = headState.GetBeaconProposerIndexForSlot(targetSlot)
+ require.NoError(t, err)
+
+ advanced, err := headState.Copy()
+ require.NoError(t, err)
+ require.NoError(t, transition.DefaultMachine.ProcessSlots(advanced, targetSlot))
+ newProposer, err = advanced.GetBeaconProposerIndexForSlot(targetSlot)
+ require.NoError(t, err)
+ if oldProposer != newProposer {
+ found = true
+ break
+ }
+ if nonce == 255 {
+ break
+ }
+ }
+ require.True(t, found, "fixture must expose a proposer change across epoch processing")
+ validatorParams.SetFeeRecipient(newProposer, common.Address{0x11})
+
+ baseBlockRoot := common.Hash{0x41}
+ syncedDataMock := syncedData.(*sync_mock_services.MockSyncedData)
+ syncedDataMock.EXPECT().ViewHeadStateWithIdentity(gomock.Any()).
+ DoAndReturn(func(view synced_data.ViewHeadStateWithIdentityFn) error {
+ return view(headState, baseBlockRoot, headState.Slot())
+ })
+ syncedDataMock.EXPECT().SelectedHead().Return(baseBlockRoot, headState.Slot(), true)
+ engine := execution_client.NewMockExecutionEngine(ctrl)
+ engine.EXPECT().ForkChoiceUpdate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
+ DoAndReturn(func(_ context.Context, _, _, _ common.Hash, attrs *engine_types.PayloadAttributes, _ clparams.StateVersion) ([]byte, error) {
+ require.Equal(t, common.Address{0x11}, attrs.SuggestedFeeRecipient)
+ return []byte{1, 2, 3, 4, 5, 6, 7, 8}, nil
+ })
+ handler.engine = engine
+
+ clock := eth_clock.NewMockEthereumClock(ctrl)
+ clock.EXPECT().GetSlotTime(gomock.Any()).Return(time.Now().Add(6 * time.Second)).AnyTimes()
+ handler.ethClock = clock
+
+ _, err := handler.preparePayloadFor(t.Context(), targetSlot)
+ require.NoError(t, err)
+}
+
+func TestPreparePayloadForPairsRootAndStateFromOneView(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ _, _, _, _, postState, handler, _, syncedData, _, validatorParams := setupTestingHandler(t, clparams.ElectraVersion, log.Root(), false)
+ targetSlot := postState.Slot() + 1
+ proposerIndex, err := postState.GetBeaconProposerIndexForSlot(targetSlot)
+ require.NoError(t, err)
+ validatorParams.SetFeeRecipient(proposerIndex, common.Address{0x11})
+
+ // The root the view hands over is the only one preparation may use. Reading it separately would
+ // let a head update in between pair a parent beacon block root with a different state, priming a
+ // builder production can never match.
+ viewRoot := common.Hash{0x41}
+ syncedDataMock := syncedData.(*sync_mock_services.MockSyncedData)
+ syncedDataMock.EXPECT().ViewHeadStateWithIdentity(gomock.Any()).
+ DoAndReturn(func(view synced_data.ViewHeadStateWithIdentityFn) error {
+ return view(postState, viewRoot, postState.Slot())
+ })
+ syncedDataMock.EXPECT().SelectedHead().Return(viewRoot, postState.Slot(), true)
+
+ clock := eth_clock.NewMockEthereumClock(ctrl)
+ clock.EXPECT().GetSlotTime(gomock.Any()).Return(time.Now().Add(6 * time.Second)).AnyTimes()
+ handler.ethClock = clock
+
+ engine := execution_client.NewMockExecutionEngine(ctrl)
+ engine.EXPECT().ForkChoiceUpdate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
+ DoAndReturn(func(_ context.Context, _, _, _ common.Hash, attrs *engine_types.PayloadAttributes, _ clparams.StateVersion) ([]byte, error) {
+ require.NotNil(t, attrs.ParentBeaconBlockRoot)
+ require.Equal(t, viewRoot, *attrs.ParentBeaconBlockRoot)
+ return []byte{1, 2, 3, 4, 5, 6, 7, 8}, nil
+ })
+ handler.engine = engine
+
+ primedHead, err := handler.preparePayloadFor(t.Context(), targetSlot)
+ require.NoError(t, err)
+ require.Equal(t, viewRoot, primedHead)
+}
+
+func TestGetEthV3ValidatorBlockMapsNotSyncedToServiceUnavailable(t *testing.T) {
+ _, _, _, _, _, handler, _, syncedData, _, _ := setupTestingHandler(t, clparams.ElectraVersion, log.Root(), false)
+ syncedDataMock := syncedData.(*sync_mock_services.MockSyncedData)
+ syncedDataMock.EXPECT().ViewHeadStateWithIdentity(gomock.Any()).Return(synced_data.ErrNotSynced)
+
+ req := httptest.NewRequest(http.MethodGet, "/?randao_reveal="+hexutil.Encode(make([]byte, 96)), nil)
+ routeCtx := chi.NewRouteContext()
+ routeCtx.URLParams.Add("slot", "1")
+ req = req.WithContext(context.WithValue(req.Context(), chi.RouteCtxKey, routeCtx))
+
+ _, err := handler.GetEthV3ValidatorBlock(httptest.NewRecorder(), req)
+ var endpointErr *beaconhttp.EndpointError
+ require.ErrorAs(t, err, &endpointErr)
+ require.Equal(t, http.StatusServiceUnavailable, endpointErr.Code)
+}
+
+func TestShouldPreparePayloadVersion(t *testing.T) {
+ for _, tc := range []struct {
+ version clparams.StateVersion
+ want bool
+ }{
+ {clparams.Phase0Version, false},
+ {clparams.AltairVersion, false},
+ {clparams.BellatrixVersion, false},
+ {clparams.CapellaVersion, true},
+ {clparams.DenebVersion, true},
+ {clparams.FuluVersion, true},
+ {clparams.GloasVersion, false},
+ } {
+ require.Equal(t, tc.want, shouldPreparePayloadVersion(tc.version), tc.version.String())
+ }
+}
+
+func TestPayloadAttributesCarryEveryFieldTheExecutionLayerGates(t *testing.T) {
+ root := common.Hash{0xaa}
+ withdrawals := []*types.Withdrawal{{Index: 1}}
+
+ attrs := payloadAttributes(1, common.Hash{0xbb}, common.Address{0xcc}, withdrawals, &root)
+
+ require.Equal(t, withdrawals, attrs.Withdrawals)
+ require.Equal(t, &root, attrs.ParentBeaconBlockRoot)
+ require.Nil(t, attrs.SlotNumber, "only a Gloas proposal knows its slot number")
+ require.Nil(t, attrs.TargetGasLimit)
+}
+
+func TestPreparedPayloadKeepsConsecutiveSlots(t *testing.T) {
+ var p preparedPayload
+ first := []byte{1, 1, 1, 1, 1, 1, 1, 1}
+ second := []byte{2, 2, 2, 2, 2, 2, 2, 2}
+ now := time.Unix(100, 0)
+
+ // Consecutive proposals: priming slot 11 must not evict slot 10, whose block may still be
+ // in production.
+ p.set(10, first, now)
+ p.set(11, second, now)
+ require.True(t, p.matches(10, first, now, 0))
+ require.True(t, p.matches(11, second, now, 0))
+
+ // Records old enough that they can no longer be produced are dropped, so the map is bounded.
+ p.set(10+preparedPayloadRetainSlots+1, []byte{3, 3, 3, 3, 3, 3, 3, 3}, now)
+ require.False(t, p.matches(10, first, now, 0))
+}
+
+func TestPreparedPayloadCopiesTheID(t *testing.T) {
+ var p preparedPayload
+ id := []byte{1, 2, 3, 4, 5, 6, 7, 8}
+ now := time.Unix(100, 0)
+
+ p.set(10, id, now)
+ id[0] = 0xff
+
+ // The caller's buffer must not be able to invalidate, or forge, a later match.
+ require.True(t, p.matches(10, []byte{1, 2, 3, 4, 5, 6, 7, 8}, now, 0))
+ require.False(t, p.matches(10, id, now, 0))
+}
+
+func TestSlotHeadMatchesOnlyTheExactPairing(t *testing.T) {
+ headA := common.Hash{0xaa}
+ headB := common.Hash{0xbb}
+ settled := slotHead{slot: 10, head: headA, generation: 3}
+
+ // Nothing recorded yet matches nothing.
+ require.NotEqual(t, settled, slotHead{})
+
+ // Already settled for this slot, on this head, under these registrations: nothing to do.
+ require.Equal(t, settled, slotHead{slot: 10, head: headA, generation: 3})
+
+ // The previous slot's block arrived late and moved the head, so the warm builder is on a parent
+ // that is no longer the head — prime again rather than wait for the next slot.
+ require.NotEqual(t, settled, slotHead{slot: 10, head: headB, generation: 3})
+
+ // A new target slot always needs priming.
+ require.NotEqual(t, settled, slotHead{slot: 11, head: headA, generation: 3})
+
+ // A registration arriving late can make a slot ours after it was ruled out, or change the fee
+ // recipient a prime already went out with.
+ require.NotEqual(t, settled, slotHead{slot: 10, head: headA, generation: 4})
+}
diff --git a/cl/beacon/handler/handler.go b/cl/beacon/handler/handler.go
index 1f15ca474cb..11b3261df9c 100644
--- a/cl/beacon/handler/handler.go
+++ b/cl/beacon/handler/handler.go
@@ -106,6 +106,13 @@ type ApiHandler struct {
routerCfg *beacon_router_configuration.RouterConfiguration
logger log.Logger
+ // preparedPayload tracks the payload primed ahead of a slot this node proposes.
+ preparedPayload preparedPayload
+ // proposalsInFlight counts block productions running on this node, so preparation can stand
+ // off. Both go through the execution layer's weight-one semaphore, and a prime holding it
+ // across the collection window turns a fully built payload into a missed slot.
+ proposalsInFlight atomic.Int64
+
// Validator data structures
validatorParams *validator_params.ValidatorParams
blobBundles *lru.Cache[common.Bytes48, BlobBundle] // Keep recent bundled blobs from the execution layer.
diff --git a/cl/beacon/handler/payload_preparation.go b/cl/beacon/handler/payload_preparation.go
new file mode 100644
index 00000000000..8c93899fa64
--- /dev/null
+++ b/cl/beacon/handler/payload_preparation.go
@@ -0,0 +1,372 @@
+// Copyright 2026 The Erigon Authors
+// This file is part of Erigon.
+//
+// Erigon is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// Erigon is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with Erigon. If not, see .
+
+package handler
+
+import (
+ "bytes"
+ "context"
+ "errors"
+ "sync"
+ "time"
+
+ "github.com/erigontech/erigon/cl/beacon/synced_data"
+ "github.com/erigontech/erigon/cl/clparams"
+ "github.com/erigontech/erigon/cl/cltypes"
+ "github.com/erigontech/erigon/cl/phase1/core/state"
+ "github.com/erigontech/erigon/cl/phase1/execution_client"
+ "github.com/erigontech/erigon/cl/transition"
+ "github.com/erigontech/erigon/common"
+ "github.com/erigontech/erigon/common/hexutil"
+ "github.com/erigontech/erigon/common/log/v3"
+ "github.com/erigontech/erigon/execution/engineapi/engine_types"
+ "github.com/erigontech/erigon/execution/execmodule/chainreader"
+)
+
+var (
+ errNotOurProposal = errors.New("next slot is not proposed by a registered validator")
+ errNoPayloadID = errors.New("execution layer returned no payload id")
+ errHeadTooFarBack = errors.New("head state is too far behind the slot to prepare")
+ errPreparationHeadChanged = errors.New("head changed while preparing payload")
+ errPreparationTooLate = errors.New("slot is too close to prime a payload production would use")
+)
+
+// preparedPayloadRetainSlots keeps a primed record alive past the slot it was primed for, so
+// priming the next slot cannot evict the record for a proposal that is still being produced.
+const preparedPayloadRetainSlots = 2
+
+type preparedPayloadRecord struct {
+ id []byte
+ primedAt time.Time
+}
+
+type preparedPayload struct {
+ mu sync.Mutex
+ payloads map[uint64]preparedPayloadRecord
+}
+
+func (p *preparedPayload) set(slot uint64, payloadID []byte, primedAt time.Time) {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ if p.payloads == nil {
+ p.payloads = map[uint64]preparedPayloadRecord{}
+ }
+ // Slots this far back can no longer be produced, so dropping them bounds the map.
+ for recorded := range p.payloads {
+ if recorded+preparedPayloadRetainSlots < slot {
+ delete(p.payloads, recorded)
+ }
+ }
+ p.payloads[slot] = preparedPayloadRecord{id: bytes.Clone(payloadID), primedAt: primedAt}
+}
+
+func (p *preparedPayload) matches(slot uint64, payloadID []byte, now time.Time, minAge time.Duration) bool {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ record, ok := p.payloads[slot]
+ return ok && len(payloadID) > 0 && bytes.Equal(record.id, payloadID) && now.Sub(record.primedAt) >= minAge
+}
+
+func canUsePreparedPayload(p *preparedPayload, builderContinuity bool, slot uint64, payloadID []byte, now time.Time, minAge time.Duration) bool {
+ return builderContinuity && p.matches(slot, payloadID, now, minAge)
+}
+
+// StartPayloadPreparation primes the execution layer for slots this node is due to propose, so the
+// payload is already packed when the validator client asks for a block instead of being built from
+// scratch inside the proposal slot.
+func (a *ApiHandler) StartPayloadPreparation(ctx context.Context) {
+ if a.routerCfg == nil || !a.routerCfg.Validator || a.engine == nil || !a.engine.SupportInsertion() {
+ return
+ }
+ go a.preparePayloadLoop(ctx)
+}
+
+func (a *ApiHandler) preparePayloadLoop(ctx context.Context) {
+ // A quarter-slot tick lands well inside every slot without assuming when in the slot the head
+ // arrives; preparation is skipped unless the next slot is ours, so the cost is a proposer
+ // lookup on a state we already hold.
+ tick := time.Duration(a.beaconChainCfg.SecondsPerSlot) * time.Second / 4
+ // Preparation is silent on a node that rarely proposes, so say once that it is running:
+ // otherwise a loop that never started looks exactly like one with nothing to do.
+ log.Info("PayloadPreparation: watching for proposals", "every", tick)
+ ticker := time.NewTicker(tick)
+ defer ticker.Stop()
+
+ var (
+ primed slotHead
+ notOurs slotHead
+ lastFailureLog time.Time
+ )
+ for immediate := true; ; immediate = false {
+ if immediate {
+ select {
+ case <-ctx.Done():
+ return
+ default:
+ }
+ } else {
+ select {
+ case <-ctx.Done():
+ return
+ case <-ticker.C:
+ }
+ }
+
+ targetSlot := a.ethClock.GetCurrentSlot() + 1
+ stateVersion := a.beaconChainCfg.GetCurrentStateVersion(targetSlot / a.beaconChainCfg.SlotsPerEpoch)
+ if preparationRetired(stateVersion) {
+ return
+ }
+ if !shouldPreparePayloadVersion(stateVersion) {
+ continue
+ }
+ // On consecutive proposals the prime for the next slot lands inside this one.
+ if a.proposalsInFlight.Load() > 0 {
+ continue
+ }
+ // Nothing is registered, so nothing here can be ours. Checking first keeps a non-validating
+ // node off the state copy entirely.
+ generation := a.validatorParams.Generation()
+ if generation == 0 {
+ continue
+ }
+ // Before genesis the current slot clamps to zero, so the next slot can be arbitrarily far
+ // off; a builder primed that early hits its own cap before the slot even starts. Too late
+ // and the prime can never reach the age production demands of it.
+ slotStart := a.ethClock.GetSlotTime(targetSlot)
+ lead := time.Until(slotStart)
+ if lead > maxPreparationLead(a.beaconChainCfg) || lead < preparedPayloadMinimumAge(a.beaconChainCfg, stateVersion) {
+ continue
+ }
+ selectedRoot, _, selected := a.syncedData.SelectedHead()
+ if !selected {
+ continue
+ }
+ // Fork choice publishes the selected head before the execution layer is notified and
+ // before the memoized state catches up. Priming while those disagree would send
+ // attributes for a head the execution layer has already moved past, unwinding the
+ // block it just executed right before this node proposes.
+ if selectedRoot != a.syncedData.HeadRoot() {
+ continue
+ }
+ // Both verdicts survive only while the slot, its parent head and the registrations they
+ // were taken under all hold. Re-deriving the proposer costs a state copy, and an epoch
+ // transition on top when the slot is in the next epoch.
+ settled := slotHead{slot: targetSlot, head: selectedRoot, generation: generation}
+ if primed == settled || notOurs == settled {
+ continue
+ }
+ prepareCtx, cancel := context.WithDeadline(ctx, slotStart)
+ head, err := a.preparePayloadFor(prepareCtx, targetSlot)
+ cancel()
+ if err != nil {
+ if errors.Is(err, errNotOurProposal) {
+ notOurs = settled
+ }
+ if !isExpectedPreparationSkip(err) && time.Since(lastFailureLog) >= time.Minute {
+ log.Warn("PayloadPreparation: failed", "slot", targetSlot, "err", err)
+ lastFailureLog = time.Now()
+ }
+ continue
+ }
+ primed = slotHead{slot: targetSlot, head: head, generation: generation}
+ }
+}
+
+// slotHead records work already settled for a target slot, on a given head, under a given set of
+// validator registrations. A zero value matches nothing reachable: preparation never runs before
+// the first registration arrives.
+type slotHead struct {
+ slot uint64
+ head common.Hash
+ generation uint64
+}
+
+// maxPreparationLead bounds how far ahead of a slot priming is worthwhile. One slot is all a live
+// chain ever offers, since preparation only ever targets the slot after the current one.
+func maxPreparationLead(cfg *clparams.BeaconChainConfig) time.Duration {
+ return time.Duration(cfg.SecondsPerSlot) * time.Second
+}
+
+// preparationRetired is the single authority for the fork after which builders gossip bids
+// instead of being primed, so the loop and the per-slot check cannot drift apart.
+func preparationRetired(version clparams.StateVersion) bool {
+ return version.AfterOrEqual(clparams.GloasVersion)
+}
+
+// shouldPreparePayloadVersion starts at Capella because payload attributes always carry withdrawals,
+// which the execution layer rejects before Shanghai: priming earlier could only ever fail.
+func shouldPreparePayloadVersion(version clparams.StateVersion) bool {
+ return version.AfterOrEqual(clparams.CapellaVersion) && !preparationRetired(version)
+}
+
+// isExpectedPreparationSkip reports whether there was simply nothing to prepare, as opposed to a
+// failure worth reporting.
+func isExpectedPreparationSkip(err error) bool {
+ return errors.Is(err, errNotOurProposal) ||
+ errors.Is(err, errNoPayloadID) ||
+ errors.Is(err, errHeadTooFarBack) ||
+ errors.Is(err, errPreparationHeadChanged) ||
+ errors.Is(err, errPreparationTooLate) ||
+ errors.Is(err, execution_client.ErrForkChoiceNotAdopted) ||
+ errors.Is(err, execution_client.ErrForkChoiceBusy) ||
+ errors.Is(err, chainreader.ErrExecutionBusy) ||
+ errors.Is(err, context.DeadlineExceeded) ||
+ errors.Is(err, context.Canceled) ||
+ errors.Is(err, synced_data.ErrNotSynced)
+}
+
+// preparePayloadFor sends the forkchoice update for targetSlot ahead of the slot itself.
+func (a *ApiHandler) preparePayloadFor(ctx context.Context, targetSlot uint64) (common.Hash, error) {
+ var (
+ baseBlockRoot common.Hash
+ proposerIndex uint64
+ feeRecipient common.Address
+ baseState *state.CachingBeaconState
+ lookupAfterAdvance bool
+ )
+ // Root, proposer and state all come from one view of the head. Reading them separately would
+ // let a head update in between pair a parent beacon block root with a different state, priming
+ // a builder that production can never match.
+ if err := a.syncedData.ViewHeadStateWithIdentity(func(headState *state.CachingBeaconState, root common.Hash, _ uint64) error {
+ baseBlockRoot = root
+ // Beyond the proposer lookahead the index has to be reshuffled from the seed, which is far
+ // too costly to repeat every tick on a large validator set.
+ slotsPerEpoch := a.beaconChainCfg.SlotsPerEpoch
+ if targetSlot/slotsPerEpoch > headState.Slot()/slotsPerEpoch+a.beaconChainCfg.MinSeedLookahead {
+ return errHeadTooFarBack
+ }
+
+ lookupAfterAdvance = targetSlot/slotsPerEpoch > headState.Slot()/slotsPerEpoch
+ var err error
+ if !lookupAfterAdvance {
+ if proposerIndex, err = headState.GetBeaconProposerIndexForSlot(targetSlot); err != nil {
+ return err
+ }
+ var ok bool
+ if feeRecipient, ok = a.validatorParams.GetFeeRecipient(proposerIndex); !ok {
+ return errNotOurProposal
+ }
+ }
+ baseState, err = headState.Copy()
+ return err
+ }); err != nil {
+ return common.Hash{}, err
+ }
+
+ if err := transition.DefaultMachine.ProcessSlots(baseState, targetSlot); err != nil {
+ return common.Hash{}, err
+ }
+ if lookupAfterAdvance {
+ var err error
+ if proposerIndex, err = baseState.GetBeaconProposerIndexForSlot(targetSlot); err != nil {
+ return common.Hash{}, err
+ }
+ var ok bool
+ if feeRecipient, ok = a.validatorParams.GetFeeRecipient(proposerIndex); !ok {
+ return common.Hash{}, errNotOurProposal
+ }
+ }
+
+ stateVersion := a.beaconChainCfg.GetCurrentStateVersion(targetSlot / a.beaconChainCfg.SlotsPerEpoch)
+ // The lead was measured before the copy and the epoch transition, which are slow enough to have
+ // spent it. Stop rather than record a prime production would reject on age.
+ if time.Until(a.ethClock.GetSlotTime(targetSlot)) < preparedPayloadMinimumAge(a.beaconChainCfg, stateVersion) {
+ return common.Hash{}, errPreparationTooLate
+ }
+ head, safeHash, finalizedHash, attrs, err := a.preGloasForkChoiceInputs(baseState, baseBlockRoot, targetSlot, feeRecipient)
+ if err != nil {
+ return common.Hash{}, err
+ }
+ payloadID, err := a.forkChoiceUpdateForPreparation(ctx, baseBlockRoot, finalizedHash, safeHash, head, attrs, stateVersion)
+ if err != nil {
+ return common.Hash{}, err
+ }
+ if len(payloadID) == 0 {
+ return common.Hash{}, errNoPayloadID
+ }
+
+ a.preparedPayload.set(targetSlot, payloadID, time.Now())
+ log.Info("PayloadPreparation: primed execution layer", "slot", targetSlot, "proposer", proposerIndex, "head", baseBlockRoot)
+ return baseBlockRoot, nil
+}
+
+// forkChoiceUpdateForPreparation retries contention within the prime, whose context already ends at
+// the slot it is for. Everything upstream of it — the head view, the state copy, the epoch
+// transition — is far more expensive to repeat than the update itself. The head is re-checked before
+// every attempt so a retry can never assert one fork choice has already left behind.
+func (a *ApiHandler) forkChoiceUpdateForPreparation(
+ ctx context.Context,
+ baseBlockRoot common.Hash,
+ finalized, safe, head common.Hash,
+ attrs *engine_types.PayloadAttributes,
+ stateVersion clparams.StateVersion,
+) ([]byte, error) {
+ for {
+ if selectedRoot, _, selected := a.syncedData.SelectedHead(); !selected || selectedRoot != baseBlockRoot {
+ return nil, errPreparationHeadChanged
+ }
+ payloadID, err := a.engine.ForkChoiceUpdate(ctx, finalized, safe, head, attrs, stateVersion)
+ if !errors.Is(err, execution_client.ErrForkChoiceBusy) {
+ return payloadID, err
+ }
+ select {
+ case <-ctx.Done():
+ return nil, ctx.Err()
+ case <-time.After(forkChoiceBusyRetryDelay):
+ }
+ }
+}
+
+// executionCheckpointHashes resolves the safe and finalized execution hashes, falling back to the
+// head for a checkpoint whose execution block this node has not seen yet.
+func (a *ApiHandler) executionCheckpointHashes(baseState *state.CachingBeaconState, head common.Hash) (safeHash, finalizedHash common.Hash) {
+ finalizedHash = a.forkchoiceStore.GetFinalizedExecutionHash(baseState.FinalizedCheckpoint().Root)
+ if finalizedHash == (common.Hash{}) {
+ finalizedHash = head
+ }
+ safeHash = a.forkchoiceStore.GetFinalizedExecutionHash(baseState.CurrentJustifiedCheckpoint().Root)
+ if safeHash == (common.Hash{}) {
+ safeHash = head
+ }
+ return safeHash, finalizedHash
+}
+
+// preGloasForkChoiceInputs builds the shared preparation and production inputs.
+func (a *ApiHandler) preGloasForkChoiceInputs(
+ baseState *state.CachingBeaconState,
+ baseBlockRoot common.Hash,
+ targetSlot uint64,
+ feeRecipient common.Address,
+) (head, safeHash, finalizedHash common.Hash, attrs *engine_types.PayloadAttributes, err error) {
+ head = baseState.LatestExecutionPayloadHeader().BlockHash
+ safeHash, finalizedHash = a.executionCheckpointHashes(baseState, head)
+
+ epoch := targetSlot / a.beaconChainCfg.SlotsPerEpoch
+ clWithdrawals, err := state.GetExpectedWithdrawals(baseState, epoch)
+ if err != nil {
+ return head, safeHash, finalizedHash, nil, err
+ }
+
+ attrs = payloadAttributes(
+ hexutil.Uint64(state.ComputeTimestampAtSlot(baseState, targetSlot)),
+ baseState.GetRandaoMixes(epoch),
+ feeRecipient,
+ cltypes.ConvertConsensusWithdrawalsToExecutionWithdrawals(clWithdrawals.Withdrawals),
+ &baseBlockRoot,
+ )
+ return head, safeHash, finalizedHash, attrs, nil
+}
diff --git a/cl/beacon/handler/utils_test.go b/cl/beacon/handler/utils_test.go
index 71657e54917..ae00500217e 100644
--- a/cl/beacon/handler/utils_test.go
+++ b/cl/beacon/handler/utils_test.go
@@ -87,8 +87,14 @@ func setupTestingHandler(t *testing.T, v clparams.StateVersion, logger log.Logge
bcfg.InitializeForkSchedule()
if useRealSyncDataMgr {
- syncedData = synced_data.NewSyncedDataManager(&bcfg, true)
- syncedData.OnHeadState(postState)
+ manager := synced_data.NewSyncedDataManager(&bcfg, true)
+ manager.OnHeadState(postState)
+ // Fork choice publishes the selected head before the state is memoized, so a synced
+ // fixture has to expose both for handlers that read head identity.
+ headRoot, err := postState.BlockRoot()
+ require.NoError(t, err)
+ manager.OnSelectedHead(headRoot, postState.Slot())
+ syncedData = manager
} else {
syncedData = sync_mock_services.NewMockSyncedData(ctrl)
}
diff --git a/cl/cltypes/withdrawal.go b/cl/cltypes/withdrawal.go
index 40447af3601..d20bdb7cb71 100644
--- a/cl/cltypes/withdrawal.go
+++ b/cl/cltypes/withdrawal.go
@@ -87,6 +87,16 @@ func convertExecutionWithdrawalsToConsensusWithdrawals(executionWithdrawal []*ty
return ret
}
+// ConvertConsensusWithdrawalsToExecutionWithdrawals is the single crossing point between the two
+// representations, so payload attributes built on different code paths cannot drift apart.
+func ConvertConsensusWithdrawalsToExecutionWithdrawals(consensusWithdrawals []*Withdrawal) []*types.Withdrawal {
+ ret := make([]*types.Withdrawal, len(consensusWithdrawals))
+ for i, w := range consensusWithdrawals {
+ ret[i] = convertConsensusWithdrawalToExecutionWithdrawal(w)
+ }
+ return ret
+}
+
// ExpectedWithdrawals represents the expected withdrawals for a beacon state
type ExpectedWithdrawals struct {
Withdrawals []*Withdrawal `json:"withdrawals"`
diff --git a/cl/phase1/execution_client/execution_client_direct.go b/cl/phase1/execution_client/execution_client_direct.go
index ab5e5a9290f..0f8984f4d96 100644
--- a/cl/phase1/execution_client/execution_client_direct.go
+++ b/cl/phase1/execution_client/execution_client_direct.go
@@ -29,6 +29,7 @@ import (
"github.com/erigontech/erigon/cl/monitor"
"github.com/erigontech/erigon/common"
"github.com/erigontech/erigon/common/hexutil"
+ "github.com/erigontech/erigon/common/log/v3"
"github.com/erigontech/erigon/execution/engineapi/engine_types"
"github.com/erigontech/erigon/execution/execmodule"
"github.com/erigontech/erigon/execution/execmodule/chainreader"
@@ -126,6 +127,30 @@ func (cc *ExecutionClientDirect) NewPayload(
return PayloadStatusNone, errors.New("unexpected status")
}
+// ErrForkChoiceNotAdopted reports that the execution layer did not adopt the requested head, so
+// there is nothing to build on.
+var ErrForkChoiceNotAdopted = errors.New("execution layer did not adopt forkchoice head")
+
+// ErrForkChoiceBusy reports contention rather than rejection. The execution layer either declined
+// the update outright or is still running it in the background; the two are indistinguishable from
+// here, and in both cases only a later attempt settles it. Retrying is the caller's decision,
+// because only the caller knows whether the head it asked for is still the one it wants.
+var ErrForkChoiceBusy = errors.New("execution layer busy with a forkchoice update")
+
+// forkChoiceStatusError classifies a status reached with payload attributes attached. Only an
+// adopted head is safe to build on: a builder ignores the parent it was asked for and packs on top
+// of whatever the execution tip really is.
+func forkChoiceStatusError(status execmodule.ExecutionStatus) error {
+ switch status {
+ case execmodule.ExecutionStatusSuccess:
+ return nil
+ case execmodule.ExecutionStatusBusy:
+ return ErrForkChoiceBusy
+ default:
+ return fmt.Errorf("%w: status %d", ErrForkChoiceNotAdopted, status)
+ }
+}
+
func (cc *ExecutionClientDirect) ForkChoiceUpdate(ctx context.Context, finalized, safe, head common.Hash, attr *engine_types.PayloadAttributes, _ clparams.StateVersion) ([]byte, error) {
status, _, _, err := cc.chainRW.UpdateForkChoice(ctx, head, safe, finalized)
if err != nil {
@@ -138,20 +163,21 @@ func (cc *ExecutionClientDirect) ForkChoiceUpdate(ctx context.Context, finalized
return nil, errors.New("bad block as forkchoice")
}
if attr == nil {
+ if status == execmodule.ExecutionStatusBusy {
+ log.Debug("[ForkChoiceUpdated] execution layer busy, head may not have been applied", "head", head)
+ }
return nil, nil
}
+ if err := forkChoiceStatusError(status); err != nil {
+ return nil, err
+ }
// Retry AssembleBlock if the EL is busy (semaphore contention with
// fork choice commits). This is common in single-process dev mode
// where the CL and EL share the same process.
idBytes := make([]byte, 8)
- var id uint64
- for range 30 {
- id, err = cc.chainRW.AssembleBlock(head, attr)
- if err == nil {
- break
- }
- time.Sleep(200 * time.Millisecond)
- }
+ id, err := retryAssembleBlock(ctx, 30, 200*time.Millisecond, func(ctx context.Context) (uint64, error) {
+ return cc.chainRW.AssembleBlock(ctx, head, attr)
+ })
if err != nil {
return nil, err
}
@@ -159,6 +185,43 @@ func (cc *ExecutionClientDirect) ForkChoiceUpdate(ctx context.Context, finalized
return idBytes, nil
}
+func retryAssembleBlock(ctx context.Context, attempts int, delay time.Duration, assemble func(context.Context) (uint64, error)) (uint64, error) {
+ if attempts <= 0 {
+ return 0, errors.New("assemble block requires at least one attempt")
+ }
+ var (
+ id uint64
+ err error
+ )
+ for attempt := range attempts {
+ if ctxErr := ctx.Err(); ctxErr != nil {
+ return 0, ctxErr
+ }
+ if id, err = assemble(ctx); err == nil {
+ return id, nil
+ }
+ if !errors.Is(err, chainreader.ErrExecutionBusy) {
+ return 0, err
+ }
+ if attempt+1 == attempts {
+ break
+ }
+ timer := time.NewTimer(delay)
+ select {
+ case <-ctx.Done():
+ if !timer.Stop() {
+ select {
+ case <-timer.C:
+ default:
+ }
+ }
+ return 0, ctx.Err()
+ case <-timer.C:
+ }
+ }
+ return 0, err
+}
+
func (cc *ExecutionClientDirect) SupportInsertion() bool {
return true
}
@@ -202,8 +265,8 @@ func (cc *ExecutionClientDirect) HasBlock(ctx context.Context, hash common.Hash)
return cc.chainRW.HasBlock(ctx, hash)
}
-func (cc *ExecutionClientDirect) GetAssembledBlock(_ context.Context, idBytes []byte, _ clparams.StateVersion) (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error) {
- return cc.chainRW.GetAssembledBlock(binary.LittleEndian.Uint64(idBytes))
+func (cc *ExecutionClientDirect) GetAssembledBlock(ctx context.Context, idBytes []byte, _ clparams.StateVersion) (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error) {
+ return cc.chainRW.GetAssembledBlock(ctx, binary.LittleEndian.Uint64(idBytes))
}
func (cc *ExecutionClientDirect) HasGapInSnapshots(ctx context.Context) bool {
diff --git a/cl/phase1/execution_client/execution_client_direct_test.go b/cl/phase1/execution_client/execution_client_direct_test.go
new file mode 100644
index 00000000000..7556411f58a
--- /dev/null
+++ b/cl/phase1/execution_client/execution_client_direct_test.go
@@ -0,0 +1,106 @@
+// Copyright 2026 The Erigon Authors
+// This file is part of Erigon.
+//
+// Erigon is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// Erigon is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with Erigon. If not, see .
+
+package execution_client
+
+import (
+ "context"
+ "errors"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/require"
+
+ "github.com/erigontech/erigon/execution/execmodule"
+ "github.com/erigontech/erigon/execution/execmodule/chainreader"
+)
+
+func TestRetryAssembleBlockStopsWhenContextIsCanceled(t *testing.T) {
+ ctx, cancel := context.WithCancel(t.Context())
+ calls := 0
+ _, err := retryAssembleBlock(ctx, 30, time.Hour, func(context.Context) (uint64, error) {
+ calls++
+ cancel()
+ return 0, chainreader.ErrExecutionBusy
+ })
+
+ require.ErrorIs(t, err, context.Canceled)
+ require.Equal(t, 1, calls)
+}
+
+func TestRetryAssembleBlockDoesNotStartWithCanceledContext(t *testing.T) {
+ ctx, cancel := context.WithCancel(t.Context())
+ cancel()
+ calls := 0
+ _, err := retryAssembleBlock(ctx, 30, time.Hour, func(context.Context) (uint64, error) {
+ calls++
+ return 0, chainreader.ErrExecutionBusy
+ })
+
+ require.ErrorIs(t, err, context.Canceled)
+ require.Zero(t, calls)
+}
+
+func TestRetryAssembleBlockStopsOnPermanentError(t *testing.T) {
+ rejected := errors.New("withdrawals before shanghai")
+ calls := 0
+ _, err := retryAssembleBlock(t.Context(), 30, time.Hour, func(context.Context) (uint64, error) {
+ calls++
+ return 0, rejected
+ })
+
+ require.ErrorIs(t, err, rejected)
+ require.Equal(t, 1, calls)
+}
+
+func TestRetryAssembleBlockReturnsFirstSuccess(t *testing.T) {
+ calls := 0
+ id, err := retryAssembleBlock(t.Context(), 3, time.Millisecond, func(context.Context) (uint64, error) {
+ calls++
+ if calls < 3 {
+ return 0, chainreader.ErrExecutionBusy
+ }
+ return 7, nil
+ })
+
+ require.NoError(t, err)
+ require.Equal(t, uint64(7), id)
+ require.Equal(t, 3, calls)
+}
+
+func TestRetryAssembleBlockRejectsNoAttempts(t *testing.T) {
+ _, err := retryAssembleBlock(t.Context(), 0, time.Millisecond, func(context.Context) (uint64, error) {
+ return 1, nil
+ })
+ require.EqualError(t, err, "assemble block requires at least one attempt")
+}
+
+func TestForkChoiceStatusErrors(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ status execmodule.ExecutionStatus
+ want error
+ }{
+ {"busy is contention, not rejection", execmodule.ExecutionStatusBusy, ErrForkChoiceBusy},
+ {"too far away", execmodule.ExecutionStatusTooFarAway, ErrForkChoiceNotAdopted},
+ {"missing segment", execmodule.ExecutionStatusMissingSegment, ErrForkChoiceNotAdopted},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ require.ErrorIs(t, forkChoiceStatusError(tc.status), tc.want)
+ })
+ }
+ require.NoError(t, forkChoiceStatusError(execmodule.ExecutionStatusSuccess))
+}
diff --git a/cl/phase1/execution_client/execution_client_engine.go b/cl/phase1/execution_client/execution_client_engine.go
index d3ef2ad26cc..3715a335f41 100644
--- a/cl/phase1/execution_client/execution_client_engine.go
+++ b/cl/phase1/execution_client/execution_client_engine.go
@@ -320,7 +320,7 @@ func (cc *ExecutionClientEngine) HasBlock(ctx context.Context, hash common.Hash)
func (cc *ExecutionClientEngine) GetAssembledBlock(ctx context.Context, id []byte, version clparams.StateVersion) (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error) {
if cc.isLocal() {
- return cc.chainRW.GetAssembledBlock(binary.LittleEndian.Uint64(id))
+ return cc.chainRW.GetAssembledBlock(ctx, binary.LittleEndian.Uint64(id))
}
// GetPayload versions advance with the response fields introduced by each fork.
diff --git a/cl/phase1/stages/forkchoice.go b/cl/phase1/stages/forkchoice.go
index abfd0a9b6f7..b8651cbe1e2 100644
--- a/cl/phase1/stages/forkchoice.go
+++ b/cl/phase1/stages/forkchoice.go
@@ -14,6 +14,7 @@ import (
"github.com/erigontech/erigon/cl/beacon/beaconevents"
"github.com/erigontech/erigon/cl/beacon/synced_data"
"github.com/erigontech/erigon/cl/clparams"
+ "github.com/erigontech/erigon/cl/cltypes"
"github.com/erigontech/erigon/cl/monitor"
"github.com/erigontech/erigon/cl/monitor/shuffling_metrics"
"github.com/erigontech/erigon/cl/persistence/beacon_indicies"
@@ -30,7 +31,6 @@ import (
"github.com/erigontech/erigon/db/datadir"
"github.com/erigontech/erigon/db/kv"
"github.com/erigontech/erigon/execution/engineapi/engine_types"
- "github.com/erigontech/erigon/execution/types"
)
// computeAndNotifyServicesOfNewForkChoice calculates the new head of the fork choice and notifies relevant services.
@@ -251,19 +251,11 @@ func emitNextPaylodAttributesEvent(cfg *Cfg, headSlot uint64, headRoot common.Ha
log.Warn("failed to get proposer index", "err", err)
return err
}
- withdrawals := []*types.Withdrawal{}
expWithdrawals, err := state.GetExpectedWithdrawals(s, epoch)
if err != nil {
return err
}
- for _, w := range expWithdrawals.Withdrawals {
- withdrawals = append(withdrawals, &types.Withdrawal{
- Amount: w.Amount,
- Index: w.Index,
- Validator: w.Validator,
- Address: w.Address,
- })
- }
+ withdrawals := cltypes.ConvertConsensusWithdrawalsToExecutionWithdrawals(expWithdrawals.Withdrawals)
payloadAttributes := engine_types.PayloadAttributes{
Timestamp: hexutil.Uint64(headPayloadHeader.Time + cfg.beaconCfg.SecondsPerSlot),
PrevRandao: randaoMix,
diff --git a/cl/validator/validator_params/validator_params.go b/cl/validator/validator_params/validator_params.go
index f422714b8bf..3a063721bab 100644
--- a/cl/validator/validator_params/validator_params.go
+++ b/cl/validator/validator_params/validator_params.go
@@ -18,12 +18,14 @@ package validator_params
import (
"sync"
+ "sync/atomic"
"github.com/erigontech/erigon/common"
)
type ValidatorParams struct {
feeRecipients sync.Map
+ generation atomic.Uint64
}
func NewValidatorParams() *ValidatorParams {
@@ -31,7 +33,18 @@ func NewValidatorParams() *ValidatorParams {
}
func (vp *ValidatorParams) SetFeeRecipient(validatorIndex uint64, feeRecipient common.Address) {
- vp.feeRecipients.Store(validatorIndex, feeRecipient)
+ // Validator clients re-register unchanged on a schedule, so only a real change counts: a
+ // generation that moved on every call would be useless to anyone caching a lookup.
+ if previous, loaded := vp.feeRecipients.Swap(validatorIndex, feeRecipient); !loaded || previous.(common.Address) != feeRecipient {
+ vp.generation.Add(1)
+ }
+}
+
+// Generation changes whenever a registration is added or its fee recipient changes, and is zero
+// until the first one arrives. A consumer caching a lookup can compare it to find out whether a
+// later registration could have changed the answer.
+func (vp *ValidatorParams) Generation() uint64 {
+ return vp.generation.Load()
}
func (vp *ValidatorParams) GetFeeRecipient(validatorIndex uint64) (common.Address, bool) {
diff --git a/cmd/caplin/caplin1/run.go b/cmd/caplin/caplin1/run.go
index a744357bc8e..91aa5ac5bfc 100644
--- a/cmd/caplin/caplin1/run.go
+++ b/cmd/caplin/caplin1/run.go
@@ -623,6 +623,7 @@ func RunCaplinService(ctx context.Context, engine execution_client.ExecutionEngi
payloadAttestationService,
proposerPreferencesService,
)
+ apiHandler.StartPayloadPreparation(ctx)
go beacon.ListenAndServe(&beacon.LayeredBeaconHandler{
ArchiveApi: apiHandler,
}, config.BeaconAPIRouter)
diff --git a/execution/builder/block_builder.go b/execution/builder/block_builder.go
index e77019a08c2..2deb3b43897 100644
--- a/execution/builder/block_builder.go
+++ b/execution/builder/block_builder.go
@@ -89,7 +89,7 @@ func NewBlockBuilder(build BlockBuilderFunc, param *Parameters, maxBuildTime tim
}
func (b *BlockBuilder) Stop(ctx context.Context) (*types.BlockWithReceipts, error) {
- b.interrupt.Store(true)
+ b.Cancel()
select {
case <-ctx.Done():
@@ -102,6 +102,24 @@ func (b *BlockBuilder) Stop(ctx context.Context) (*types.BlockWithReceipts, erro
return b.result, b.err
}
+func (b *BlockBuilder) Cancel() {
+ b.interrupt.Store(true)
+}
+
+// Failed reports whether the builder finished without producing anything. The error is latched, so
+// a caller that would otherwise reuse this builder has to treat it as absent. Being cancelled is
+// not failure: a stopped builder still holds the payload it was stopped for.
+func (b *BlockBuilder) Failed() bool {
+ select {
+ case <-b.done:
+ default:
+ return false
+ }
+ b.mu.Lock()
+ defer b.mu.Unlock()
+ return b.err != nil
+}
+
func (b *BlockBuilder) Block() *types.Block {
b.mu.Lock()
defer b.mu.Unlock()
diff --git a/execution/builder/block_builder_test.go b/execution/builder/block_builder_test.go
new file mode 100644
index 00000000000..e0bf5aa3470
--- /dev/null
+++ b/execution/builder/block_builder_test.go
@@ -0,0 +1,83 @@
+// Copyright 2026 The Erigon Authors
+// This file is part of Erigon.
+//
+// Erigon is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// Erigon is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with Erigon. If not, see .
+
+package builder
+
+import (
+ "errors"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/require"
+
+ "github.com/erigontech/erigon/execution/types"
+)
+
+func TestBlockBuilderRunningHasNotFailed(t *testing.T) {
+ t.Parallel()
+
+ release := make(chan struct{})
+ t.Cleanup(func() { close(release) })
+ b := NewBlockBuilder(func(_ *Parameters, _ *atomic.Bool) (*types.BlockWithReceipts, error) {
+ <-release
+ return nil, errors.New("builder stopped")
+ }, &Parameters{}, time.Minute)
+
+ require.Never(t, b.Failed, 50*time.Millisecond, 5*time.Millisecond)
+}
+
+func TestBlockBuilderStoppedForItsPayloadHasNotFailed(t *testing.T) {
+ t.Parallel()
+
+ b := NewBlockBuilder(func(_ *Parameters, interrupt *atomic.Bool) (*types.BlockWithReceipts, error) {
+ for !interrupt.Load() {
+ time.Sleep(time.Millisecond)
+ }
+ return &types.BlockWithReceipts{Block: types.NewBlock(&types.Header{}, nil, nil, nil, nil)}, nil
+ }, &Parameters{}, time.Minute)
+
+ _, err := b.Stop(t.Context())
+ require.NoError(t, err)
+
+ // Collecting the payload is what a proposal does. Reading that as failure would make a repeated
+ // request rebuild from scratch instead of being handed the block that was just built.
+ require.False(t, b.Failed())
+}
+
+func TestBlockBuilderHasFailedOnceItErrors(t *testing.T) {
+ t.Parallel()
+
+ b := NewBlockBuilder(func(_ *Parameters, _ *atomic.Bool) (*types.BlockWithReceipts, error) {
+ return nil, errors.New("build failed")
+ }, &Parameters{}, time.Minute)
+
+ require.Eventually(t, b.Failed, time.Second, time.Millisecond)
+}
+
+func TestBlockBuilderStaysReusableOnceItFillsTheBlock(t *testing.T) {
+ t.Parallel()
+
+ built := make(chan struct{})
+ b := NewBlockBuilder(func(_ *Parameters, _ *atomic.Bool) (*types.BlockWithReceipts, error) {
+ defer close(built)
+ return &types.BlockWithReceipts{Block: types.NewBlock(&types.Header{}, nil, nil, nil, nil)}, nil
+ }, &Parameters{}, time.Minute)
+
+ <-built
+ // A builder that ran out of room holds a complete payload, so its id is still worth reusing.
+ require.Never(t, b.Failed, 50*time.Millisecond, 5*time.Millisecond)
+}
diff --git a/execution/execmodule/block_building.go b/execution/execmodule/block_building.go
index ef1d94cddaf..7eb38e84d67 100644
--- a/execution/execmodule/block_building.go
+++ b/execution/execmodule/block_building.go
@@ -17,6 +17,7 @@
package execmodule
import (
+ "bytes"
"context"
"reflect"
"time"
@@ -40,14 +41,10 @@ func (e *ExecModule) checkWithdrawalsPresence(time uint64, withdrawals []*types.
return nil
}
-// buildDuration returns how long a payload builder may run before it stops itself.
-//
-// A consensus layer may send payload attributes well ahead of the slot the payload is for and then
-// call getPayload against the cached payload id, without sending fresh attributes. A builder that
-// stops before that slot hands back a payload missing every transaction that arrived in between, so
-// build until shortly into the target slot, derived from the payload's own timestamp. The floor
-// leaves a late request no worse off than a fixed budget; the cap stops an implausible timestamp
-// from pinning a builder and its resources.
+// buildDuration spans the target slot for early requests while bounding late and implausibly
+// future requests. A consensus layer may send attributes well ahead of the slot and then call
+// getPayload against the cached id without sending fresh ones, so a builder that stopped before
+// the slot would hand back a payload missing every transaction that arrived in between.
func buildDuration(payloadTimestamp uint64, now time.Time, secondsPerSlot uint64) time.Duration {
slot := time.Duration(secondsPerSlot) * time.Second
// Reject beyond the cap horizon before converting: a large enough timestamp overflows
@@ -60,16 +57,75 @@ func buildDuration(payloadTimestamp uint64, now time.Time, secondsPerSlot uint64
return min(max(d, slot/4), 2*slot)
}
+func cloneBuilderParameters(params *builder.Parameters) *builder.Parameters {
+ if params == nil {
+ return nil
+ }
+ cloned := *params
+ cloned.ExtraData = bytes.Clone(params.ExtraData)
+ if params.Withdrawals != nil {
+ cloned.Withdrawals = make([]*types.Withdrawal, len(params.Withdrawals))
+ for i, withdrawal := range params.Withdrawals {
+ if withdrawal != nil {
+ copy := *withdrawal
+ cloned.Withdrawals[i] = ©
+ }
+ }
+ }
+ if params.ParentBeaconBlockRoot != nil {
+ copy := *params.ParentBeaconBlockRoot
+ cloned.ParentBeaconBlockRoot = ©
+ }
+ if params.SlotNumber != nil {
+ copy := *params.SlotNumber
+ cloned.SlotNumber = ©
+ }
+ if params.TargetGasLimit != nil {
+ copy := *params.TargetGasLimit
+ cloned.TargetGasLimit = ©
+ }
+ return &cloned
+}
+
+// builderEntry keeps a builder with the parameters and timestamp it was created for, so the
+// three cannot drift apart and eviction can drop the timestamp index without scanning it.
+type builderEntry struct {
+ builder *builder.BlockBuilder
+ params *builder.Parameters
+ timestamp uint64
+}
+
+func (e *ExecModule) dropBuilder(id uint64, entry *builderEntry) {
+ if e.buildersByTimestamp[entry.timestamp] == id {
+ delete(e.buildersByTimestamp, entry.timestamp)
+ }
+ delete(e.builders, id)
+}
+
func (e *ExecModule) evictOldBuilders() {
ids := common.SortedKeys(e.builders)
// remove old builders so that at most MaxBuilders - 1 remain
for i := 0; i <= len(e.builders)-engine_helpers.MaxBuilders; i++ {
- delete(e.builders, ids[i])
+ id := ids[i]
+ if old := e.builders[id]; old != nil {
+ if old.builder != nil {
+ old.builder.Cancel()
+ }
+ if e.buildersByTimestamp[old.timestamp] == id {
+ delete(e.buildersByTimestamp, old.timestamp)
+ }
+ }
+ delete(e.builders, id)
}
}
func (e *ExecModule) AssembleBlock(ctx context.Context, params *builder.Parameters) (AssembleBlockResult, error) {
+ // Cancellation is checked first so an expired request reports why it stopped instead of
+ // masquerading as contention, which callers retry.
+ if err := ctx.Err(); err != nil {
+ return AssembleBlockResult{}, err
+ }
if !e.semaphore.TryAcquire(1) {
return AssembleBlockResult{Busy: true}, nil
}
@@ -79,23 +135,35 @@ func (e *ExecModule) AssembleBlock(ctx context.Context, params *builder.Paramete
return AssembleBlockResult{}, err
}
- // First check if we're already building a block with the requested parameters
- if e.lastParameters != nil {
- params.PayloadId = e.lastParameters.PayloadId
- if reflect.DeepEqual(e.lastParameters, params) {
- e.logger.Info("[ForkChoiceUpdated] duplicate build request")
- return AssembleBlockResult{PayloadID: e.lastParameters.PayloadId}, nil
+ // A stopped builder is still worth reusing: it holds the payload it was stopped for, which is
+ // exactly what a repeated request is asking for. Only a failed one has to be passed over.
+ if previousID, ok := e.buildersByTimestamp[params.Timestamp]; ok {
+ if previous := e.builders[previousID]; previous != nil && previous.builder != nil && !previous.builder.Failed() {
+ params.PayloadId = previousID
+ if reflect.DeepEqual(previous.params, params) {
+ e.logger.Info("[ForkChoiceUpdated] duplicate build request")
+ return AssembleBlockResult{PayloadID: previousID}, nil
+ }
}
}
-
- // Initiate payload building
+ // A superseded builder keeps running to its own deadline. The timestamp index moves to the new
+ // one, so nothing reaches it by dedup, while an id already handed out goes on answering with a
+ // payload that is still growing.
e.evictOldBuilders()
e.nextPayloadId++
params.PayloadId = e.nextPayloadId
- e.lastParameters = params
+ ownedParams := cloneBuilderParameters(params)
- e.builders[e.nextPayloadId] = builder.NewBlockBuilder(e.builderFunc, params, buildDuration(params.Timestamp, time.Now(), e.config.SecondsPerSlot()))
+ if e.buildersByTimestamp == nil {
+ e.buildersByTimestamp = make(map[uint64]uint64)
+ }
+ e.builders[e.nextPayloadId] = &builderEntry{
+ builder: builder.NewBlockBuilder(e.builderFunc, ownedParams, buildDuration(params.Timestamp, time.Now(), e.config.SecondsPerSlot())),
+ params: ownedParams,
+ timestamp: params.Timestamp,
+ }
+ e.buildersByTimestamp[params.Timestamp] = e.nextPayloadId
e.logger.Info("[ForkChoiceUpdated] BlockBuilder added", "payload", e.nextPayloadId)
return AssembleBlockResult{PayloadID: e.nextPayloadId}, nil
@@ -118,17 +186,25 @@ func blockValue(br *types.BlockWithReceipts, baseFee *uint256.Int) *uint256.Int
}
func (e *ExecModule) GetAssembledBlock(ctx context.Context, payloadID uint64) (AssembledBlockResult, error) {
+ if err := ctx.Err(); err != nil {
+ return AssembledBlockResult{}, err
+ }
if !e.semaphore.TryAcquire(1) {
return AssembledBlockResult{Busy: true}, nil
}
defer e.semaphore.Release(1)
- bldr, ok := e.builders[payloadID]
- if !ok {
+ entry, ok := e.builders[payloadID]
+ if !ok || entry == nil || entry.builder == nil {
return AssembledBlockResult{}, nil
}
- blockWithReceipts, err := bldr.Stop(ctx)
+ blockWithReceipts, err := entry.builder.Stop(ctx)
if err != nil {
+ // Keeping a failed entry would hand the same latched error to every retry. A caller whose
+ // own context expired says nothing about the builder.
+ if ctx.Err() == nil {
+ e.dropBuilder(payloadID, entry)
+ }
e.logger.Error("Failed to build PoS block", "err", err)
return AssembledBlockResult{}, err
}
diff --git a/execution/execmodule/block_building_internal_test.go b/execution/execmodule/block_building_internal_test.go
index b48479af2e6..7d85b5eca16 100644
--- a/execution/execmodule/block_building_internal_test.go
+++ b/execution/execmodule/block_building_internal_test.go
@@ -17,13 +17,353 @@
package execmodule
import (
+ "context"
+ "errors"
"math"
+ "sync/atomic"
"testing"
"time"
+ "golang.org/x/sync/semaphore"
+
"github.com/stretchr/testify/require"
+
+ "github.com/erigontech/erigon/common"
+ "github.com/erigontech/erigon/common/log/v3"
+ "github.com/erigontech/erigon/execution/builder"
+ "github.com/erigontech/erigon/execution/chain"
+ "github.com/erigontech/erigon/execution/engineapi/engine_helpers"
+ "github.com/erigontech/erigon/execution/types"
)
+func TestAssembleBlockKeepsBuildersApartByTimestamp(t *testing.T) {
+ type runningBuilder struct {
+ id uint64
+ interrupt *atomic.Bool
+ }
+ started := make(chan runningBuilder, 4)
+ module := &ExecModule{
+ semaphore: semaphore.NewWeighted(1),
+ config: &chain.Config{},
+ logger: log.Root(),
+ builders: map[uint64]*builderEntry{},
+ builderFunc: func(params *builder.Parameters, interrupt *atomic.Bool) (*types.BlockWithReceipts, error) {
+ started <- runningBuilder{id: params.PayloadId, interrupt: interrupt}
+ for !interrupt.Load() {
+ time.Sleep(time.Millisecond)
+ }
+ return nil, errors.New("builder stopped")
+ },
+ }
+ t.Cleanup(func() {
+ for _, entry := range module.builders {
+ if entry != nil && entry.builder != nil {
+ _, _ = entry.builder.Stop(context.Background())
+ }
+ }
+ })
+
+ waitStarted := func() runningBuilder {
+ t.Helper()
+ select {
+ case running := <-started:
+ return running
+ case <-time.After(time.Second):
+ t.Fatal("builder did not start")
+ return runningBuilder{}
+ }
+ }
+ assemble := func(timestamp uint64, parent common.Hash) (uint64, runningBuilder) {
+ t.Helper()
+ result, err := module.AssembleBlock(t.Context(), &builder.Parameters{Timestamp: timestamp, ParentHash: parent})
+ require.NoError(t, err)
+ require.False(t, result.Busy)
+ return result.PayloadID, waitStarted()
+ }
+
+ firstID, first := assemble(100, common.Hash{0x01})
+ adjacentID, adjacent := assemble(101, common.Hash{0x02})
+ require.NotEqual(t, firstID, adjacentID)
+
+ firstDuplicate, err := module.AssembleBlock(t.Context(), &builder.Parameters{Timestamp: 100, ParentHash: common.Hash{0x01}})
+ require.NoError(t, err)
+ require.Equal(t, firstID, firstDuplicate.PayloadID)
+
+ // Superseding hands the timestamp to a new builder and leaves the old ones running, so only the
+ // index moves. Timestamp 101 is a different proposal and is untouched throughout.
+ secondID, second := assemble(100, common.Hash{0x03})
+ require.NotEqual(t, firstID, secondID)
+ require.Equal(t, secondID, module.buildersByTimestamp[100])
+ require.Equal(t, adjacentID, module.buildersByTimestamp[101])
+
+ thirdID, third := assemble(100, common.Hash{0x04})
+ require.NotEqual(t, secondID, thirdID)
+ require.Equal(t, thirdID, module.buildersByTimestamp[100])
+
+ duplicate, err := module.AssembleBlock(t.Context(), &builder.Parameters{Timestamp: 100, ParentHash: common.Hash{0x04}})
+ require.NoError(t, err)
+ require.Equal(t, thirdID, duplicate.PayloadID)
+
+ for _, running := range []runningBuilder{first, second, third, adjacent} {
+ require.False(t, running.interrupt.Load(), "builder %d must still be packing", running.id)
+ }
+
+ // Eviction is where a builder is actually stopped, and it takes the timestamp index with it.
+ delete(module.builders, firstID)
+ delete(module.builders, secondID)
+ for id := thirdID + 1; len(module.builders) < engine_helpers.MaxBuilders; id++ {
+ module.builders[id] = nil
+ }
+ module.evictOldBuilders()
+ require.Eventually(t, adjacent.interrupt.Load, time.Second, time.Millisecond)
+ require.NotContains(t, module.builders, adjacentID)
+ require.NotContains(t, module.buildersByTimestamp, uint64(101))
+ require.False(t, third.interrupt.Load(), "the current builder for a timestamp must survive eviction")
+}
+
+func TestSupersededBuilderKeepsPackingAndStaysRetrievable(t *testing.T) {
+ started := make(chan *atomic.Bool, 4)
+ module := &ExecModule{
+ logger: log.Root(),
+ config: &chain.Config{},
+ semaphore: semaphore.NewWeighted(1),
+ builders: map[uint64]*builderEntry{},
+ builderFunc: func(_ *builder.Parameters, interrupt *atomic.Bool) (*types.BlockWithReceipts, error) {
+ started <- interrupt
+ for !interrupt.Load() {
+ time.Sleep(time.Millisecond)
+ }
+ return &types.BlockWithReceipts{Block: types.NewBlock(&types.Header{}, nil, nil, nil, nil)}, nil
+ },
+ }
+
+ first, err := module.AssembleBlock(t.Context(), &builder.Parameters{Timestamp: 100, ParentHash: common.Hash{0x01}})
+ require.NoError(t, err)
+ firstInterrupt := <-started
+
+ second, err := module.AssembleBlock(t.Context(), &builder.Parameters{Timestamp: 100, ParentHash: common.Hash{0x02}})
+ require.NoError(t, err)
+ require.NotEqual(t, first.PayloadID, second.PayloadID)
+ secondInterrupt := <-started
+
+ // The timestamp index moves to the new builder, so nothing reaches the old one by dedup. It is
+ // left running: freezing it would answer an id already handed out with a near-empty payload.
+ require.Equal(t, second.PayloadID, module.buildersByTimestamp[100])
+ require.False(t, firstInterrupt.Load())
+
+ assembled, err := module.GetAssembledBlock(t.Context(), first.PayloadID)
+ require.NoError(t, err)
+ require.NotNil(t, assembled.Block)
+
+ secondInterrupt.Store(true)
+}
+
+func TestCollectedPayloadIsHandedBackToARepeatedRequest(t *testing.T) {
+ started := make(chan struct{}, 4)
+ module := &ExecModule{
+ logger: log.Root(),
+ config: &chain.Config{},
+ semaphore: semaphore.NewWeighted(1),
+ builders: map[uint64]*builderEntry{},
+ builderFunc: func(_ *builder.Parameters, interrupt *atomic.Bool) (*types.BlockWithReceipts, error) {
+ started <- struct{}{}
+ for !interrupt.Load() {
+ time.Sleep(time.Millisecond)
+ }
+ return &types.BlockWithReceipts{Block: types.NewBlock(&types.Header{}, nil, nil, nil, nil)}, nil
+ },
+ }
+
+ params := func() *builder.Parameters {
+ return &builder.Parameters{Timestamp: 100, ParentHash: common.Hash{0x01}}
+ }
+ first, err := module.AssembleBlock(t.Context(), params())
+ require.NoError(t, err)
+ <-started
+
+ assembled, err := module.GetAssembledBlock(t.Context(), first.PayloadID)
+ require.NoError(t, err)
+ require.NotNil(t, assembled.Block)
+
+ // Collecting stops the builder. A repeated request must still be handed that payload: rebuilding
+ // from scratch this late means the next grab takes a near-empty block.
+ repeat, err := module.AssembleBlock(t.Context(), params())
+ require.NoError(t, err)
+ require.Equal(t, first.PayloadID, repeat.PayloadID)
+ require.Empty(t, started, "a repeated request must not start a second builder")
+}
+
+func TestAssembleBlockDoesNotReuseFailedBuilder(t *testing.T) {
+ var failNext atomic.Bool
+ failNext.Store(true)
+ started := make(chan struct{}, 4)
+ module := &ExecModule{
+ logger: log.Root(),
+ config: &chain.Config{},
+ semaphore: semaphore.NewWeighted(1),
+ builders: map[uint64]*builderEntry{},
+ builderFunc: func(_ *builder.Parameters, interrupt *atomic.Bool) (*types.BlockWithReceipts, error) {
+ started <- struct{}{}
+ if failNext.Swap(false) {
+ return nil, errors.New("build failed")
+ }
+ for !interrupt.Load() {
+ time.Sleep(time.Millisecond)
+ }
+ return nil, errors.New("builder stopped")
+ },
+ }
+
+ params := func() *builder.Parameters {
+ return &builder.Parameters{Timestamp: 100, ParentHash: common.Hash{0x01}}
+ }
+ first, err := module.AssembleBlock(t.Context(), params())
+ require.NoError(t, err)
+ <-started
+ require.Eventually(t, module.builders[first.PayloadID].builder.Failed, time.Second, time.Millisecond)
+
+ // Identical parameters would normally dedup onto the same id. A builder that already died
+ // latches its error, so reusing it would spend the slot on a payload that can never arrive.
+ second, err := module.AssembleBlock(t.Context(), params())
+ require.NoError(t, err)
+ require.NotEqual(t, first.PayloadID, second.PayloadID)
+ <-started
+
+ _, _ = module.builders[second.PayloadID].builder.Stop(context.Background())
+}
+
+func TestGetAssembledBlockDropsFailedBuilder(t *testing.T) {
+ module := &ExecModule{
+ logger: log.Root(),
+ config: &chain.Config{},
+ semaphore: semaphore.NewWeighted(1),
+ builders: map[uint64]*builderEntry{},
+ builderFunc: func(_ *builder.Parameters, _ *atomic.Bool) (*types.BlockWithReceipts, error) {
+ return nil, errors.New("build failed")
+ },
+ }
+
+ result, err := module.AssembleBlock(t.Context(), &builder.Parameters{Timestamp: 100, ParentHash: common.Hash{0x01}})
+ require.NoError(t, err)
+
+ _, err = module.GetAssembledBlock(t.Context(), result.PayloadID)
+ require.Error(t, err)
+
+ // The error is latched, so leaving the entry in place would keep serving it to every retry.
+ require.NotContains(t, module.builders, result.PayloadID)
+ require.NotContains(t, module.buildersByTimestamp, uint64(100))
+}
+
+func TestAssembleBlockOwnsParameters(t *testing.T) {
+ type observedParameters struct {
+ parentRoot common.Hash
+ extraData byte
+ }
+ readParameters := make(chan struct{})
+ observed := make(chan observedParameters, 1)
+ module := &ExecModule{
+ semaphore: semaphore.NewWeighted(1),
+ config: &chain.Config{},
+ logger: log.Root(),
+ builders: map[uint64]*builderEntry{},
+ builderFunc: func(params *builder.Parameters, interrupt *atomic.Bool) (*types.BlockWithReceipts, error) {
+ <-readParameters
+ observed <- observedParameters{parentRoot: *params.ParentBeaconBlockRoot, extraData: params.ExtraData[0]}
+ for !interrupt.Load() {
+ time.Sleep(time.Millisecond)
+ }
+ return nil, errors.New("builder stopped")
+ },
+ }
+ root := common.Hash{0xaa}
+ params := &builder.Parameters{
+ Timestamp: 100,
+ ParentHash: common.Hash{0x01},
+ ParentBeaconBlockRoot: &root,
+ ExtraData: []byte{0xbb},
+ }
+ result, err := module.AssembleBlock(t.Context(), params)
+ require.NoError(t, err)
+ require.False(t, result.Busy)
+
+ root[0] = 0xcc
+ params.ExtraData[0] = 0xdd
+ close(readParameters)
+ require.Equal(t, observedParameters{parentRoot: common.Hash{0xaa}, extraData: 0xbb}, <-observed)
+
+ duplicate, err := module.AssembleBlock(t.Context(), &builder.Parameters{
+ Timestamp: 100,
+ ParentHash: common.Hash{0x01},
+ ParentBeaconBlockRoot: &common.Hash{0xaa},
+ ExtraData: []byte{0xbb},
+ })
+ require.NoError(t, err)
+ require.Equal(t, result.PayloadID, duplicate.PayloadID)
+ _, _ = module.builders[result.PayloadID].builder.Stop(context.Background())
+}
+
+func TestAssembleBlockCanceledContextDoesNotSupersedeBuilder(t *testing.T) {
+ started := make(chan *atomic.Bool, 1)
+ module := &ExecModule{
+ semaphore: semaphore.NewWeighted(1),
+ config: &chain.Config{},
+ logger: log.Root(),
+ builders: map[uint64]*builderEntry{},
+ builderFunc: func(_ *builder.Parameters, interrupt *atomic.Bool) (*types.BlockWithReceipts, error) {
+ started <- interrupt
+ for !interrupt.Load() {
+ time.Sleep(time.Millisecond)
+ }
+ return nil, errors.New("builder stopped")
+ },
+ }
+ result, err := module.AssembleBlock(t.Context(), &builder.Parameters{Timestamp: 100, ParentHash: common.Hash{0x01}})
+ require.NoError(t, err)
+ interrupt := <-started
+ t.Cleanup(func() {
+ _, _ = module.builders[result.PayloadID].builder.Stop(context.Background())
+ })
+
+ ctx, cancel := context.WithCancel(t.Context())
+ cancel()
+ _, err = module.AssembleBlock(ctx, &builder.Parameters{Timestamp: 100, ParentHash: common.Hash{0x02}})
+ require.ErrorIs(t, err, context.Canceled)
+ require.False(t, interrupt.Load())
+ require.Equal(t, result.PayloadID, module.buildersByTimestamp[100])
+}
+
+func TestCloneBuilderParametersPreservesRepresentations(t *testing.T) {
+ require.Nil(t, cloneBuilderParameters(nil))
+
+ empty := cloneBuilderParameters(&builder.Parameters{Withdrawals: []*types.Withdrawal{}, ExtraData: []byte{}})
+ require.NotNil(t, empty.Withdrawals)
+ require.NotNil(t, empty.ExtraData)
+
+ root := common.Hash{0x01}
+ slot := uint64(2)
+ gasLimit := uint64(3)
+ params := &builder.Parameters{
+ Withdrawals: []*types.Withdrawal{nil, {Index: 4}},
+ ParentBeaconBlockRoot: &root,
+ SlotNumber: &slot,
+ TargetGasLimit: &gasLimit,
+ ExtraData: []byte{5},
+ }
+ cloned := cloneBuilderParameters(params)
+ params.Withdrawals[1].Index = 40
+ root[0] = 10
+ slot = 20
+ gasLimit = 30
+ params.ExtraData[0] = 50
+
+ require.Nil(t, cloned.Withdrawals[0])
+ require.Equal(t, uint64(4), cloned.Withdrawals[1].Index)
+ require.Equal(t, common.Hash{0x01}, *cloned.ParentBeaconBlockRoot)
+ require.Equal(t, uint64(2), *cloned.SlotNumber)
+ require.Equal(t, uint64(3), *cloned.TargetGasLimit)
+ require.Equal(t, byte(5), cloned.ExtraData[0])
+}
+
func TestBuildDuration(t *testing.T) {
const ethereum, gnosis = uint64(12), uint64(5)
slotStart := time.Unix(1_700_000_000, 0)
diff --git a/execution/execmodule/chainreader/chain_reader.go b/execution/execmodule/chainreader/chain_reader.go
index 6d9122aa278..162a7da5c8a 100644
--- a/execution/execmodule/chainreader/chain_reader.go
+++ b/execution/execmodule/chainreader/chain_reader.go
@@ -279,7 +279,11 @@ func (c ChainReaderWriterEth1) HasBlock(ctx context.Context, hash common.Hash) (
return c.executionModule.HasBlock(ctx, &hash, nil)
}
-func (c ChainReaderWriterEth1) AssembleBlock(baseHash common.Hash, attributes *engine_types.PayloadAttributes) (id uint64, err error) {
+// ErrExecutionBusy separates contention, which settles on its own, from a rejection that will
+// return the same answer however many times it is asked.
+var ErrExecutionBusy = errors.New("execution data is still syncing")
+
+func (c ChainReaderWriterEth1) AssembleBlock(ctx context.Context, baseHash common.Hash, attributes *engine_types.PayloadAttributes) (id uint64, err error) {
params := &builder.Parameters{
ParentHash: baseHash,
Timestamp: uint64(attributes.Timestamp),
@@ -290,23 +294,23 @@ func (c ChainReaderWriterEth1) AssembleBlock(baseHash common.Hash, attributes *e
TargetGasLimit: (*uint64)(attributes.TargetGasLimit),
ParentBeaconBlockRoot: attributes.ParentBeaconBlockRoot,
}
- result, err := c.executionModule.AssembleBlock(context.Background(), params)
+ result, err := c.executionModule.AssembleBlock(ctx, params)
if err != nil {
return 0, err
}
if result.Busy {
- return 0, errors.New("execution data is still syncing")
+ return 0, ErrExecutionBusy
}
return result.PayloadID, nil
}
-func (c ChainReaderWriterEth1) GetAssembledBlock(id uint64) (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error) {
- result, err := c.executionModule.GetAssembledBlock(context.Background(), id)
+func (c ChainReaderWriterEth1) GetAssembledBlock(ctx context.Context, id uint64) (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error) {
+ result, err := c.executionModule.GetAssembledBlock(ctx, id)
if err != nil {
return nil, nil, nil, nil, err
}
if result.Busy {
- return nil, nil, nil, nil, errors.New("execution data is still syncing")
+ return nil, nil, nil, nil, ErrExecutionBusy
}
if result.Block == nil {
return nil, nil, nil, nil, nil
diff --git a/execution/execmodule/exec_module.go b/execution/execmodule/exec_module.go
index edd986a4984..f7c8f2f7d34 100644
--- a/execution/execmodule/exec_module.go
+++ b/execution/execmodule/exec_module.go
@@ -194,10 +194,10 @@ type ExecModule struct {
logger log.Logger
// Block building
- nextPayloadId uint64
- lastParameters *builder.Parameters
- builderFunc builder.BlockBuilderFunc
- builders map[uint64]*builder.BlockBuilder
+ nextPayloadId uint64
+ builderFunc builder.BlockBuilderFunc
+ builders map[uint64]*builderEntry
+ buildersByTimestamp map[uint64]uint64
// Changes accumulator
hook *stageloop.Hook
@@ -266,7 +266,8 @@ func NewExecModule(
logger: logger,
forkValidator: forkValidator,
pipelineExecutor: pipelineExecutor,
- builders: make(map[uint64]*builder.BlockBuilder),
+ builders: make(map[uint64]*builderEntry),
+ buildersByTimestamp: make(map[uint64]uint64),
builderFunc: builderFunc,
config: config,
semaphore: semaphore.NewWeighted(1),
diff --git a/execution/execmodule/exec_module_test.go b/execution/execmodule/exec_module_test.go
index 76c4272935f..b1120584f2a 100644
--- a/execution/execmodule/exec_module_test.go
+++ b/execution/execmodule/exec_module_test.go
@@ -1276,7 +1276,7 @@ func TestAssembleBlockWithWithdrawalRequest(t *testing.T) {
time.Hour,
)
- eth1Block, blobsBundle, requestsBundle, blockValue, err := chainRW.GetAssembledBlock(payloadId)
+ eth1Block, blobsBundle, requestsBundle, blockValue, err := chainRW.GetAssembledBlock(ctx, payloadId)
require.NoError(t, err)
require.NotNil(t, eth1Block, "Eth1Block should not be nil")
require.NotNil(t, blobsBundle, "BlobsBundle should not be nil")