From 2013fa4c8470b6ea55c3295f858735303274af17 Mon Sep 17 00:00:00 2001 From: lystopad Date: Fri, 14 Aug 2026 09:28:24 +0000 Subject: [PATCH] cl/phase1, execution: give the execution module a typed busy signal and the caller's context (#23273) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Split out of #23105, which grew too large to review in one piece. Independent of the other parts of that series. ### Contention was indistinguishable from rejection `AssembleBlock` and `GetAssembledBlock` reported a busy execution module as `errors.New("execution data is still syncing")`. That message is inaccurate — it is weight-one semaphore contention with a forkchoice update or another payload request, not syncing — and being an untyped error, callers could not tell it apart from a real rejection. Both now return `chainreader.ErrExecutionBusy`, and the assemble retry waits only on that. Previously a permanent rejection — mismatched withdrawals, for instance — was retried thirty times across six seconds before surfacing, which spends the proposal slot rather than reporting the problem. ### The retry ignored its caller The loop was `for range 30 { ...; time.Sleep(200 * time.Millisecond) }`, with no context check anywhere. A cancelled caller still waited the full six seconds. It is now an extracted helper that checks the context before each attempt and waits on it rather than on a bare sleep. While there, `ChainReaderWriterEth1.AssembleBlock` and `GetAssembledBlock` take the caller's context instead of substituting `context.Background()`, so the deadline a caller sets actually reaches the execution module. ### Tests The helper is covered directly: first success, stopping on a rejection, exhausting attempts on contention, cancellation mid-flight and before the first attempt, and the zero-attempts guard. Part of a series splitting #23105. (cherry picked from commit aeb7f3bdd7f77aed8abaf2e86523d9925d8c5dfb) --- .../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 1a0b81cf98f..ead307010b9 100644 --- a/cl/phase1/execution_client/execution_client_direct.go +++ b/cl/phase1/execution_client/execution_client_direct.go @@ -139,14 +139,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 } @@ -154,6 +149,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 } @@ -197,8 +229,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 6f1aa918164..e908532333f 100644 --- a/cl/phase1/execution_client/execution_client_engine.go +++ b/cl/phase1/execution_client/execution_client_engine.go @@ -371,7 +371,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)) } // Select Engine API version based on CL state version. diff --git a/execution/execmodule/chainreader/chain_reader.go b/execution/execmodule/chainreader/chain_reader.go index 3a438ccf1dc..475298d6120 100644 --- a/execution/execmodule/chainreader/chain_reader.go +++ b/execution/execmodule/chainreader/chain_reader.go @@ -293,7 +293,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), @@ -304,23 +308,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 76a1eb48b37..96795b46111 100644 --- a/execution/execmodule/exec_module_test.go +++ b/execution/execmodule/exec_module_test.go @@ -1271,7 +1271,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")