diff --git a/cl/phase1/execution_client/execution_client_direct.go b/cl/phase1/execution_client/execution_client_direct.go index ab5e5a9290f..b020f5ec7ab 100644 --- a/cl/phase1/execution_client/execution_client_direct.go +++ b/cl/phase1/execution_client/execution_client_direct.go @@ -144,14 +144,9 @@ func (cc *ExecutionClientDirect) ForkChoiceUpdate(ctx context.Context, finalized // 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 +154,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 +234,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..6f2af144871 --- /dev/null +++ b/cl/phase1/execution_client/execution_client_direct_test.go @@ -0,0 +1,101 @@ +// 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/chainreader" +) + +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 TestRetryAssembleBlockStopsOnRejection(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 + }) + + // Only contention settles by waiting; a rejection answers the same way however often it is + // asked, so retrying it just burns the slot. + require.ErrorIs(t, err, rejected) + require.Equal(t, 1, calls) +} + +func TestRetryAssembleBlockGivesUpAfterAttempts(t *testing.T) { + calls := 0 + _, err := retryAssembleBlock(t.Context(), 2, time.Millisecond, func(context.Context) (uint64, error) { + calls++ + return 0, chainreader.ErrExecutionBusy + }) + + require.ErrorIs(t, err, chainreader.ErrExecutionBusy) + require.Equal(t, 2, calls) +} + +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 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") +} 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/execution/execmodule/chainreader/chain_reader.go b/execution/execmodule/chainreader/chain_reader.go index 6d9122aa278..cd2082c09f5 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 reports that the execution module was already occupied, which settles on its +// own, as opposed to a rejection that returns the same answer however many times it is asked. +var ErrExecutionBusy = errors.New("execution module is busy") + +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_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")