From 697974c89548a5d731860eee0245091880708c48 Mon Sep 17 00:00:00 2001 From: Oleksandr Lystopad Date: Fri, 14 Aug 2026 08:29:37 +0200 Subject: [PATCH] cl/phase1, execution: give the execution module a typed busy signal and the caller's context Split out of #23105 so it can be reviewed on its own. AssembleBlock and GetAssembledBlock reported contention as a bare error reading "execution data is still syncing", which is neither what happened nor distinguishable from a rejection. Both now return ErrExecutionBusy, and the assemble retry only waits on that: a rejection such as mismatched withdrawals used to be retried thirty times over six seconds before surfacing, which spends the slot instead of reporting it. The retry also ran with no regard for the caller, sleeping through a cancelled context for the full six seconds. It is now a helper that checks the context before every attempt and waits on it rather than on a bare sleep, and the chain reader takes the caller's context instead of substituting context.Background(). --- .../execution_client_direct.go | 52 +++++++-- .../execution_client_direct_test.go | 101 ++++++++++++++++++ .../execution_client_engine.go | 2 +- .../execmodule/chainreader/chain_reader.go | 16 +-- execution/execmodule/exec_module_test.go | 2 +- 5 files changed, 155 insertions(+), 18 deletions(-) create mode 100644 cl/phase1/execution_client/execution_client_direct_test.go 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")