Decouple retry from builder

builder/builder.go

@@ -5,7 +5,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"math/big"
 	"net/http"
 	_ "os"
 	"strconv"
@@ -60,8 +59,7 @@ type Builder struct {
 	builderPublicKey            phase0.BLSPubKey
 	builderSigningDomain        phase0.Domain
 
-	builderRetryInterval time.Duration
-	builderBlockTime     time.Duration
+	builderBlockTime time.Duration
 
 	slotMu        sync.Mutex
 	slotAttrs     BuilderPayloadAttributes
@@ -78,7 +76,6 @@ type Builder struct {
 type BuilderArgs struct {
 	sk                          *bls.SecretKey
 	builderSigningDomain        phase0.Domain
-	builderRetryInterval        time.Duration
 	blockTime                   time.Duration
 	eth                         IEthereumService
 	ignoreLatePayloadAttributes bool
@@ -115,7 +112,6 @@ func NewBuilder(args BuilderArgs) (*Builder, error) {
 		builderSecretKey:            args.sk,
 		builderPublicKey:            pk,
 		builderSigningDomain:        args.builderSigningDomain,
-		builderRetryInterval:        args.builderRetryInterval,
 		builderBlockTime:            args.blockTime,
 
 		slotCtx:       slotCtx,
@@ -396,52 +392,23 @@ func (b *Builder) handlePayloadAttributes(attrs *BuilderPayloadAttributes) error
 }
 
 func (b *Builder) runBuildingJob(slotCtx context.Context, proposerPubkey phase0.BLSPubKey, attrs *BuilderPayloadAttributes) {
-	ctx, cancel := context.WithTimeout(slotCtx, b.builderBlockTime)
-	defer cancel()
-
-	// Submission queue for the given payload attributes
-	// multiple jobs can run for different attributes fot the given slot
-	// 1. When new block is ready we check if its profit is higher than profit of last best block
-	//    if it is we set queueBest* to values of the new block and notify queueSignal channel.
-	var (
-		queueMu                sync.Mutex
-		queueLastSubmittedHash common.Hash
-		queueBestBlockValue    *big.Int = big.NewInt(0)
-	)
-
 	log.Info("runBuildingJob", "slot", attrs.Slot, "parent", attrs.HeadHash, "payloadTimestamp", uint64(attrs.Timestamp), "txs", attrs.Transactions)
 
-	// retry build block every builderBlockRetryInterval
-	runRetryLoop(ctx, b.builderRetryInterval, func() {
-		log.Info("retrying BuildBlock",
-			"slot", attrs.Slot,
-			"parent", attrs.HeadHash,
-			"retryInterval", b.builderRetryInterval.String())
-		payload, err := b.eth.BuildBlock(attrs)
-		if err != nil {
-			log.Warn("Failed to build block", "err", err)
-			return
-		}
+	payloadGenerator, err := b.eth.BuildBlock(slotCtx, attrs)
+	if err != nil {
+		log.Error("Failed to build block", "err", err)
+		return
+	}
 
-		sealedAt := time.Now()
-		queueMu.Lock()
-		defer queueMu.Unlock()
-		if payload.ExecutionPayload.BlockHash != queueLastSubmittedHash && payload.BlockValue.Cmp(queueBestBlockValue) >= 0 {
-			queueLastSubmittedHash = payload.ExecutionPayload.BlockHash
-			queueBestBlockValue = payload.BlockValue
-
-			submitBlockOpts := SubmitBlockOpts{
-				ExecutionPayloadEnvelope: payload,
-				SealedAt:                 sealedAt,
-				ProposerPubkey:           proposerPubkey,
-				PayloadAttributes:        attrs,
-			}
-			err := b.saveBlockSubmission(submitBlockOpts)
-			if err != nil {
-				log.Error("could not save block submission", "err", err)
-			}
+	for {
+		submitBlockOpts := <-payloadGenerator
+		submitBlockOpts.ProposerPubkey = proposerPubkey
+
+		err := b.saveBlockSubmission(*submitBlockOpts)
+		if err != nil {
+			log.Error("could not save block submission", "err", err)
 		}
-	})
+	}
 }
 
 func executableDataToExecutionPayload(data *engine.ExecutionPayloadEnvelope, version spec.DataVersion) (*builderApi.VersionedSubmitBlindedBlockResponse, error) {

builder/builder_test.go

@@ -1,6 +1,7 @@
 package builder
 
 import (
+	"context"
 	"math/big"
 	"testing"
 	"time"
@@ -29,8 +30,14 @@ type testEthereumService struct {
 	testBlock          *types.Block
 }
 
-func (t *testEthereumService) BuildBlock(attrs *BuilderPayloadAttributes) (*engine.ExecutionPayloadEnvelope, error) {
-	return t.testExecutableData, nil
+func (t *testEthereumService) BuildBlock(ctx context.Context, attrs *BuilderPayloadAttributes) (chan *SubmitBlockOpts, error) {
+	ch := make(chan *SubmitBlockOpts, 1)
+	ch <- &SubmitBlockOpts{ // it does not block because buffer of size 1
+		ExecutionPayloadEnvelope: t.testExecutableData,
+		SealedAt:                 time.Now(),
+		PayloadAttributes:        attrs,
+	}
+	return ch, nil
 }
 
 func (t *testEthereumService) GetBlockByHash(hash common.Hash) *types.Block { return t.testBlock }
@@ -113,7 +120,6 @@ func TestGetPayloadV1(t *testing.T) {
 	builderArgs := BuilderArgs{
 		sk:                          sk,
 		builderSigningDomain:        bDomain,
-		builderRetryInterval:        200 * time.Millisecond,
 		blockTime:                   2 * time.Second,
 		eth:                         testEthService,
 		ignoreLatePayloadAttributes: false,

builder/eth_service.go

@@ -1,7 +1,9 @@
 package builder
 
 import (
+	"context"
 	"errors"
+	"math/big"
 	"time"
 
 	"github.com/ethereum/go-ethereum/beacon/engine"
@@ -14,25 +16,74 @@ import (
 )
 
 type IEthereumService interface {
-	BuildBlock(attrs *BuilderPayloadAttributes) (*engine.ExecutionPayloadEnvelope, error)
+	BuildBlock(ctx context.Context, attrs *BuilderPayloadAttributes) (chan *SubmitBlockOpts, error)
 	GetBlockByHash(hash common.Hash) *types.Block
 	Config() *params.ChainConfig
 	Synced() bool
 }
 
 type EthereumService struct {
-	eth *eth.Ethereum
-	cfg *Config
+	eth           *eth.Ethereum
+	cfg           *Config
+	retryInterval time.Duration
 }
 
 func NewEthereumService(eth *eth.Ethereum, config *Config) *EthereumService {
 	return &EthereumService{
-		eth: eth,
-		cfg: config,
+		eth:           eth,
+		cfg:           config,
+		retryInterval: 1 * time.Second,
 	}
 }
 
-func (s *EthereumService) BuildBlock(attrs *BuilderPayloadAttributes) (*engine.ExecutionPayloadEnvelope, error) {
+func (s *EthereumService) WithRetryInterval(retryInterval time.Duration) {
+	s.retryInterval = retryInterval
+}
+
+func (s *EthereumService) BuildBlock(ctx context.Context, attrs *BuilderPayloadAttributes) (chan *SubmitBlockOpts, error) {
+	resCh := make(chan *SubmitBlockOpts, 1)
+
+	// The context already includes the timeout with the block time.
+	// Submission queue for the given payload attributes
+	// multiple jobs can run for different attributes fot the given slot
+	// 1. When new block is ready we check if its profit is higher than profit of last best block
+	//    if it is we set queueBest* to values of the new block and notify queueSignal channel.
+	var (
+		queueLastSubmittedHash common.Hash
+		queueBestBlockValue    *big.Int = big.NewInt(0)
+	)
+
+	// retry build block every builderBlockRetryInterval
+	go runRetryLoop(ctx, s.retryInterval, func() {
+		log.Info("retrying BuildBlock",
+			"slot", attrs.Slot,
+			"parent", attrs.HeadHash,
+			"retryInterval", s.retryInterval)
+
+		payload, err := s.buildBlockImpl(attrs)
+		if err != nil {
+			log.Warn("Failed to build block", "err", err)
+			return
+		}
+
+		sealedAt := time.Now()
+		if payload.ExecutionPayload.BlockHash != queueLastSubmittedHash && payload.BlockValue.Cmp(queueBestBlockValue) >= 0 {
+			queueLastSubmittedHash = payload.ExecutionPayload.BlockHash
+			queueBestBlockValue = payload.BlockValue
+
+			submitBlockOpts := SubmitBlockOpts{
+				ExecutionPayloadEnvelope: payload,
+				SealedAt:                 sealedAt,
+				PayloadAttributes:        attrs,
+			}
+			resCh <- &submitBlockOpts
+		}
+	})
+
+	return resCh, nil
+}
+
+func (s *EthereumService) buildBlockImpl(attrs *BuilderPayloadAttributes) (*engine.ExecutionPayloadEnvelope, error) {
 	// Send a request to generate a full block in the background.
 	// The result can be obtained via the returned channel.
 	args := &miner.BuildPayloadArgs{

builder/eth_service_test.go

@@ -1,6 +1,7 @@
 package builder
 
 import (
+	"context"
 	"math/big"
 	"testing"
 	"time"
@@ -92,7 +93,11 @@ func TestBuildBlock(t *testing.T) {
 
 	service := NewEthereumService(ethservice, &DefaultConfig)
 
-	executableData, err := service.BuildBlock(testPayloadAttributes)
+	payloadGenerator, err := service.BuildBlock(context.Background(), testPayloadAttributes)
+	require.NoError(t, err)
+
+	payload := <-payloadGenerator
+	executableData := payload.ExecutionPayloadEnvelope
 
 	require.Equal(t, common.Address{0x04, 0x10}, executableData.ExecutionPayload.FeeRecipient)
 	require.Equal(t, common.Hash{0x05, 0x10}, executableData.ExecutionPayload.Random)

builder/service.go

@@ -106,12 +106,12 @@ func Register(stack *node.Node, backend *eth.Ethereum, cfg *Config) error {
 	} else {
 		builderRetryInterval = RetryIntervalDefault
 	}
+	ethereumService.WithRetryInterval(builderRetryInterval)
 
 	builderArgs := BuilderArgs{
 		sk:                          builderSk,
 		eth:                         ethereumService,
 		builderSigningDomain:        builderSigningDomain,
-		builderRetryInterval:        builderRetryInterval,
 		ignoreLatePayloadAttributes: cfg.IgnoreLatePayloadAttributes,
 		beaconClient:                beaconClient,
 		blockTime:                   cfg.BlockTime,