cl/phase1, execution: give the execution module a typed busy signal and the caller's context

cl/phase1/execution_client/execution_client_direct.go

@@ -144,21 +144,53 @@ 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
 	}
 	binary.LittleEndian.PutUint64(idBytes, id)
 	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 {

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 <http://www.gnu.org/licenses/>.
+
+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")
+}

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.

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

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")