Merge pull request #158 from attestantio/parallel-proposal

Use parallel production for block proposals.
attestantio · Oct 12, 2023 · 6edeab4 · 6edeab4
2 parents ff17108 + 71f7f39
commit 6edeab4
Show file tree

Hide file tree

Showing 5 changed files with 230 additions and 89 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,6 +1,7 @@
 dev:
   - ensure relay configuration inherits all configuration values as expected
   - create strategies for builder bid
+  - fetch blinded and unblinded proposals in parallel to speed up block production
 
 1.7.6:
   - add User-Agent header to HTTP requests

diff --git a/services/beaconblockproposer/standard/propose.go b/services/beaconblockproposer/standard/propose.go
@@ -1,4 +1,4 @@
-// Copyright © 2020 - 2022 Attestant Limited.
+// Copyright © 2020 - 2023 Attestant Limited.
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
@@ -18,6 +18,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"sync"
 	"time"
 
 	"github.com/attestantio/go-block-relay/services/blockauctioneer"
@@ -67,37 +68,71 @@ func (s *Service) Propose(ctx context.Context, data interface{}) {
 		monitorBeaconBlockProposalCompleted(started, 0, s.chainTime.StartOfSlot(0), "failed")
 		return
 	}
-	if duty == nil {
-		log.Error().Msg("Passed nil data structure")
-		monitorBeaconBlockProposalCompleted(started, 0, s.chainTime.StartOfSlot(0), "failed")
+	slot, err := validateDuty(duty)
+	if err != nil {
+		log.Error().Err(err).Msg("Invalid duty")
+		monitorBeaconBlockProposalCompleted(started, slot, s.chainTime.StartOfSlot(slot), "failed")
 		return
 	}
-	span.SetAttributes(attribute.Int64("slot", int64(duty.Slot())))
-	log := log.With().Uint64("proposing_slot", uint64(duty.Slot())).Uint64("validator_index", uint64(duty.ValidatorIndex())).Logger()
+	span.SetAttributes(attribute.Int64("slot", int64(slot)))
+	log := log.With().Uint64("proposing_slot", uint64(slot)).Uint64("validator_index", uint64(duty.ValidatorIndex())).Logger()
 	log.Trace().Msg("Proposing")
 
-	var zeroSig phase0.BLSSignature
-	if duty.RANDAOReveal() == zeroSig {
-		log.Error().Msg("Missing RANDAO reveal")
-		monitorBeaconBlockProposalCompleted(started, duty.Slot(), s.chainTime.StartOfSlot(duty.Slot()), "failed")
+	graffiti, err := s.obtainGraffiti(ctx, slot, duty.ValidatorIndex())
+	if err != nil {
+		log.Warn().Err(err).Msg("Failed to obtain graffiti")
+		graffiti = nil
+	}
+
+	log.Trace().Dur("elapsed", time.Since(started)).Msg("Obtained graffiti")
+	span.AddEvent("Ready to propose")
+
+	if err := s.proposeBlock(ctx, duty, graffiti); err != nil {
+		log.Error().Err(err).Msg("Failed to propose block")
+		monitorBeaconBlockProposalCompleted(started, slot, s.chainTime.StartOfSlot(slot), "failed")
 		return
 	}
 
+	log.Trace().Dur("elapsed", time.Since(started)).Msg("Submitted proposal")
+	monitorBeaconBlockProposalCompleted(started, slot, s.chainTime.StartOfSlot(slot), "succeeded")
+}
+
+// validateDuty validates that the information supplied to us in a duty is suitable for proposing.
+func validateDuty(duty *beaconblockproposer.Duty) (phase0.Slot, error) {
+	if duty == nil {
+		return 0, errors.New("no duty supplied")
+	}
+
+	zeroSig := phase0.BLSSignature{}
+	randaoReveal := duty.RANDAOReveal()
+	if bytes.Equal(randaoReveal[:], zeroSig[:]) {
+		return duty.Slot(), errors.New("duty missing RANDAO reveal")
+	}
+
 	if duty.Account() == nil {
-		log.Error().Msg("Missing account")
-		monitorBeaconBlockProposalCompleted(started, duty.Slot(), s.chainTime.StartOfSlot(duty.Slot()), "failed")
-		return
+		return duty.Slot(), errors.New("duty missing account")
 	}
 
-	var graffiti []byte
-	var err error
-	if s.graffitiProvider != nil {
-		graffiti, err = s.graffitiProvider.Graffiti(ctx, duty.Slot(), duty.ValidatorIndex())
-		if err != nil {
-			log.Warn().Err(err).Msg("Failed to obtain graffiti")
-			graffiti = nil
-		}
+	return duty.Slot(), nil
+}
+
+// obtainGraffiti obtains the graffiti for the proposal.
+func (s *Service) obtainGraffiti(ctx context.Context,
+	slot phase0.Slot,
+	validatorIndex phase0.ValidatorIndex,
+) (
+	[]byte,
+	error,
+) {
+	if s.graffitiProvider == nil {
+		return []byte{}, nil
 	}
+
+	graffiti, err := s.graffitiProvider.Graffiti(ctx, slot, validatorIndex)
+	if err != nil {
+		return []byte{}, errors.Wrap(err, "graffiti provider failed")
+	}
+
 	if bytes.Contains(graffiti, []byte("{{CLIENT}}")) {
 		if nodeClientProvider, isProvider := s.proposalProvider.(consensusclient.NodeClientProvider); isProvider {
 			nodeClient, err := nodeClientProvider.NodeClient(ctx)
@@ -111,24 +146,31 @@ func (s *Service) Propose(ctx context.Context, data interface{}) {
 	if len(graffiti) > 32 {
 		graffiti = graffiti[0:32]
 	}
-	span.AddEvent("Ready to propose")
-	log.Trace().Dur("elapsed", time.Since(started)).Msg("Obtained graffiti")
 
-	if err := s.proposeBlock(ctx, duty, graffiti); err != nil {
-		log.Error().Err(err).Msg("Failed to propose block")
-		monitorBeaconBlockProposalCompleted(started, duty.Slot(), s.chainTime.StartOfSlot(duty.Slot()), "failed")
-		return
-	}
-
-	log.Trace().Dur("elapsed", time.Since(started)).Msg("Submitted proposal")
-	monitorBeaconBlockProposalCompleted(started, duty.Slot(), s.chainTime.StartOfSlot(duty.Slot()), "succeeded")
+	return graffiti, nil
 }
 
 // proposeBlock proposes a beacon block.
 func (s *Service) proposeBlock(ctx context.Context,
 	duty *beaconblockproposer.Duty,
 	graffiti []byte,
 ) error {
+	// Pre-fetch an unblinded block in parallel with the auction process.
+	// This ensures that we are ready to propose as quickly as possible if the auction is unsuccessful.
+	var wg sync.WaitGroup
+	var proposal *spec.VersionedBeaconBlock
+	wg.Add(1)
+	go func(ctx context.Context, duty *beaconblockproposer.Duty, graffiti []byte) {
+		var err error
+		proposal, err = s.proposalProvider.BeaconBlockProposal(ctx, duty.Slot(), duty.RANDAOReveal(), graffiti)
+		if err != nil {
+			log.Warn().Err(err).Msg("Failed to pre-obtain proposal data")
+			return
+		}
+		log.Trace().Msg("Pre-obtained proposal")
+		wg.Done()
+	}(ctx, duty, graffiti)
+
 	if s.blockAuctioneer != nil {
 		// There is a block auctioneer specified, try to propose the block with auction.
 		result := s.proposeBlockWithAuction(ctx, duty, graffiti)
@@ -145,7 +187,9 @@ func (s *Service) proposeBlock(ctx context.Context,
 		}
 	}
 
-	err := s.proposeBlockWithoutAuction(ctx, duty, graffiti)
+	wg.Wait()
+
+	err := s.proposeBlockWithoutAuction(ctx, proposal, duty, graffiti)
 	if err != nil {
 		return err
 	}
@@ -172,6 +216,7 @@ func (s *Service) proposeBlockWithAuction(ctx context.Context,
 	if auctionResults.Bid == nil {
 		return auctionResultNoBids
 	}
+	monitorBestBidRelayCount(len(auctionResults.Providers))
 
 	proposal, err := s.obtainBlindedProposal(ctx, duty, graffiti, auctionResults)
 	if err != nil {
@@ -193,7 +238,6 @@ func (s *Service) proposeBlockWithAuction(ctx context.Context,
 		log.Debug().Msg("No relays can unblind the block")
 		return auctionResultFailedCanTryWithout
 	}
-	monitorBestBidRelayCount(len(providers))
 	log.Trace().Int("providers", len(providers)).Msg("Obtained relays that can unblind the proposal")
 
 	signedBlindedBlock, err := s.signBlindedProposal(ctx, duty, proposal)
@@ -218,26 +262,29 @@ func (s *Service) proposeBlockWithAuction(ctx context.Context,
 }
 
 func (s *Service) proposeBlockWithoutAuction(ctx context.Context,
+	proposal *spec.VersionedBeaconBlock,
 	duty *beaconblockproposer.Duty,
 	graffiti []byte,
 ) error {
 	ctx, span := otel.Tracer("attestantio.vouch.services.beaconblockproposer.standard").Start(ctx, "proposeBlockWithoutAuction")
 	defer span.End()
 
-	proposal, err := s.proposalProvider.BeaconBlockProposal(ctx, duty.Slot(), duty.RANDAOReveal(), graffiti)
-	if err != nil {
-		return errors.Wrap(err, "failed to obtain proposal data")
-	}
+	var err error
 	if proposal == nil {
-		return errors.New("obtained nil beacon block proposal")
+		proposal, err = s.proposalProvider.BeaconBlockProposal(ctx, duty.Slot(), duty.RANDAOReveal(), graffiti)
+		if err != nil {
+			return errors.Wrap(err, "failed to obtain proposal data")
+		}
+		if proposal == nil {
+			return errors.New("obtained nil beacon block proposal")
+		}
+		log.Trace().Msg("Obtained proposal")
 	}
-	log.Trace().Msg("Obtained proposal")
 
 	proposalSlot, err := proposal.Slot()
 	if err != nil {
 		return errors.Wrap(err, "failed to obtain proposal slot")
 	}
-
 	if proposalSlot != duty.Slot() {
 		return errors.New("proposal data for incorrect slot")
 	}
@@ -259,7 +306,7 @@ func (s *Service) proposeBlockWithoutAuction(ctx context.Context,
 
 	sig, err := s.beaconBlockSigner.SignBeaconBlockProposal(ctx,
 		duty.Account(),
-		proposalSlot,
+		duty.Slot(),
 		duty.ValidatorIndex(),
 		parentRoot,
 		stateRoot,
@@ -269,6 +316,25 @@ func (s *Service) proposeBlockWithoutAuction(ctx context.Context,
 	}
 	log.Trace().Msg("Signed proposal")
 
+	signedBlock, err := composeVersionedSignedBeaconBlock(proposal, sig)
+	if err != nil {
+		return err
+	}
+
+	if err := s.beaconBlockSubmitter.SubmitBeaconBlock(ctx, signedBlock); err != nil {
+		return errors.Wrap(err, "failed to submit beacon block proposal")
+	}
+
+	return nil
+}
+
+func composeVersionedSignedBeaconBlock(
+	proposal *spec.VersionedBeaconBlock,
+	sig phase0.BLSSignature,
+) (
+	*spec.VersionedSignedBeaconBlock,
+	error,
+) {
 	signedBlock := &spec.VersionedSignedBeaconBlock{
 		Version: proposal.Version,
 	}
@@ -294,15 +360,10 @@ func (s *Service) proposeBlockWithoutAuction(ctx context.Context,
 			Signature: sig,
 		}
 	default:
-		return errors.New("unknown proposal version")
+		return nil, errors.New("unknown proposal version")
 	}
 
-	// Submit the block.
-	if err := s.beaconBlockSubmitter.SubmitBeaconBlock(ctx, signedBlock); err != nil {
-		return errors.Wrap(err, "failed to submit beacon block proposal")
-	}
-
-	return nil
+	return signedBlock, nil
 }
 
 func (s *Service) auctionBlock(ctx context.Context,

diff --git a/services/beaconblockproposer/standard/propose_internal_test.go b/services/beaconblockproposer/standard/propose_internal_test.go
@@ -0,0 +1,98 @@
+// Copyright © 2023 Attestant Limited.
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package standard
+
+import (
+	"context"
+	"testing"
+
+	"github.com/attestantio/go-eth2-client/spec/phase0"
+	"github.com/attestantio/vouch/services/beaconblockproposer"
+	"github.com/stretchr/testify/require"
+	e2types "github.com/wealdtech/go-eth2-types/v2"
+	keystorev4 "github.com/wealdtech/go-eth2-wallet-encryptor-keystorev4"
+	hd "github.com/wealdtech/go-eth2-wallet-hd/v2"
+	scratch "github.com/wealdtech/go-eth2-wallet-store-scratch"
+	e2wtypes "github.com/wealdtech/go-eth2-wallet-types/v2"
+)
+
+func duty(slot phase0.Slot, validatorIndex phase0.ValidatorIndex, randaoReveal phase0.BLSSignature, account e2wtypes.Account) *beaconblockproposer.Duty {
+	duty := beaconblockproposer.NewDuty(slot, validatorIndex)
+	duty.SetRandaoReveal(randaoReveal)
+	duty.SetAccount(account)
+	return duty
+}
+
+func TestValidateDuty(t *testing.T) {
+	ctx := context.Background()
+
+	// Create an account.
+	require.NoError(t, e2types.InitBLS())
+	store := scratch.New()
+	encryptor := keystorev4.New()
+	wallet, err := hd.CreateWallet(ctx, "test wallet", []byte("pass"), store, encryptor, make([]byte, 64))
+	require.NoError(t, err)
+	require.Nil(t, wallet.(e2wtypes.WalletLocker).Unlock(ctx, []byte("pass")))
+	account, err := wallet.(e2wtypes.WalletAccountCreator).CreateAccount(context.Background(), "test account", []byte("pass"))
+	require.NoError(t, err)
+	require.NoError(t, account.(e2wtypes.AccountLocker).Unlock(ctx, []byte("pass")))
+
+	sig, err := account.(e2wtypes.AccountSigner).Sign(ctx, []byte{
+		0x00, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08,
+		0x08, 0x09, 0x0a, 0x0b, 0x0c, 0x0d, 0x0e, 0x0f,
+		0x00, 0x11, 0x12, 0x13, 0x14, 0x15, 0x16, 0x17,
+		0x18, 0x19, 0x1a, 0x1b, 0x1c, 0x1d, 0x1e, 0x1f,
+	})
+	require.NoError(t, err)
+	randaoReveal := phase0.BLSSignature(sig.Marshal())
+
+	tests := []struct {
+		name string
+		duty *beaconblockproposer.Duty
+		slot phase0.Slot
+		err  string
+	}{
+		{
+			name: "Nil",
+			err:  "no duty supplied",
+		},
+		{
+			name: "NoRandaoReveal",
+			duty: duty(1, 2, phase0.BLSSignature{}, account),
+			err:  "duty missing RANDAO reveal",
+		},
+		{
+			name: "NoAccount",
+			duty: duty(1, 2, randaoReveal, nil),
+			err:  "duty missing account",
+		},
+		{
+			name: "Good",
+			duty: duty(1, 2, randaoReveal, account),
+			slot: 1,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			slot, err := validateDuty(test.duty)
+			if test.err != "" {
+				require.EqualError(t, err, test.err)
+			} else {
+				require.NoError(t, err)
+				require.Equal(t, test.slot, slot)
+			}
+		})
+	}
+}