app: handle fee recipient

app/app.go

@@ -22,10 +22,12 @@ import (
 	"crypto/ecdsa"
 	"encoding/hex"
 	"net/http"
+	"strings"
 	"time"
 
 	eth2api "github.com/attestantio/go-eth2-client/api"
 	eth2v1 "github.com/attestantio/go-eth2-client/api/v1"
+	"github.com/attestantio/go-eth2-client/spec/bellatrix"
 	eth2p0 "github.com/attestantio/go-eth2-client/spec/phase0"
 	"github.com/coinbase/kryptology/pkg/signatures/bls/bls_sig"
 	"github.com/ethereum/go-ethereum/crypto"
@@ -296,13 +298,16 @@ func wireP2P(ctx context.Context, life *lifecycle.Manager, conf Config,
 }
 
 // wireCoreWorkflow wires the core workflow components.
+//
+//nolint:gocognit
 func wireCoreWorkflow(ctx context.Context, life *lifecycle.Manager, conf Config,
 	lock cluster.Lock, nodeIdx cluster.NodeIdx, tcpNode host.Host, p2pKey *ecdsa.PrivateKey,
 	eth2Cl eth2wrap.Client, peerIDs []peer.ID, sender *p2p.Sender,
 ) error {
 	// Convert and prep public keys and public shares
 	var (
 		corePubkeys       []core.PubKey
+		eth2Pubkeys       []eth2p0.BLSPubKey
 		pubshares         []eth2p0.BLSPubKey
 		pubSharesByKey    = make(map[*bls_sig.PublicKey]*bls_sig.PublicKey)
 		allPubSharesByKey = make(map[core.PubKey]map[int]*bls_sig.PublicKey) // map[pubkey]map[shareIdx]pubshare
@@ -339,6 +344,12 @@ func wireCoreWorkflow(ctx context.Context, life *lifecycle.Manager, conf Config,
 			return err
 		}
 
+		eth2Pubkey, err := tblsconv.KeyToETH2(pubkey)
+		if err != nil {
+			return err
+		}
+
+		eth2Pubkeys = append(eth2Pubkeys, eth2Pubkey)
 		corePubkeys = append(corePubkeys, corePubkey)
 		pubSharesByKey[pubkey] = pubShare
 		pubshares = append(pubshares, eth2Share)
@@ -364,7 +375,9 @@ func wireCoreWorkflow(ctx context.Context, life *lifecycle.Manager, conf Config,
 		return err
 	}
 
-	fetch, err := fetcher.New(eth2Cl)
+	sched.SubscribeSlots(setFeeRecipient(eth2Cl, eth2Pubkeys, lock.FeeRecipientAddress))
+
+	fetch, err := fetcher.New(eth2Cl, lock.FeeRecipientAddress)
 	if err != nil {
 		return err
 	}
@@ -674,6 +687,42 @@ func wireTracing(life *lifecycle.Manager, conf Config) error {
 	return nil
 }
 
+// setFeeRecipient returns a slot subscriber for scheduler which calls prepare_beacon_proposer endpoint at start of each epoch.
+// TODO(dhruv): move this somewhere else once more use-cases like this becomes clear.
+func setFeeRecipient(eth2Cl eth2wrap.Client, pubkeys []eth2p0.BLSPubKey, feeRecipient string) func(ctx context.Context, slot core.Slot) error {
+	onStartup := true
+
+	return func(ctx context.Context, slot core.Slot) error {
+		// Either call if it is first slot in epoch or on charon startup.
+		if !onStartup && !slot.FirstInEpoch() {
+			return nil
+		}
+		onStartup = false
+
+		vals, err := eth2Cl.ValidatorsByPubKey(ctx, "head", pubkeys)
+		if err != nil {
+			return err
+		}
+
+		var addr bellatrix.ExecutionAddress
+		b, err := hex.DecodeString(strings.TrimPrefix(feeRecipient, "0x"))
+		if err != nil {
+			return errors.Wrap(err, "hex decode fee recipient address")
+		}
+		copy(addr[:], b)
+
+		var preps []*eth2v1.ProposalPreparation
+		for vIdx := range vals {
+			preps = append(preps, &eth2v1.ProposalPreparation{
+				ValidatorIndex: vIdx,
+				FeeRecipient:   addr,
+			})
+		}
+
+		return eth2Cl.SubmitProposalPreparations(ctx, preps)
+	}
+}
+
 // httpServeHook wraps a http.Server.ListenAndServe function, swallowing http.ErrServerClosed.
 type httpServeHook func() error
 

app/eth2wrap/genwrap/genwrap.go

@@ -110,6 +110,7 @@ type Client interface {
 		"NodeSyncingProvider":                   true,
 		"NodeVersionProvider":                   false,
 		"ProposerDutiesProvider":                true,
+		"ProposalPreparationsSubmitter":         false,
 		"SlotDurationProvider":                  false,
 		"SlotsPerEpochProvider":                 false,
 		"SpecProvider":                          false,

cluster/definition.go

@@ -139,7 +139,7 @@ func (d Definition) NodeIdx(pID peer.ID) (NodeIdx, error) {
 
 // VerifySignatures returns true if all config signatures are fully populated and valid. A verified definition is ready for use in DKG.
 //
-//nolint:gocognit
+
 func (d Definition) VerifySignatures() error {
 	// Skip signature verification for definition versions earlier than v1.3 since there are no EIP712 signatures before v1.3.0.
 	if !supportEIP712Sigs(d.Version) && !eip712SigsPresent(d.Operators) {

core/fetcher/fetcher.go

@@ -19,6 +19,7 @@ import (
 	"context"
 	"fmt"
 
+	"github.com/attestantio/go-eth2-client/spec"
 	eth2p0 "github.com/attestantio/go-eth2-client/spec/phase0"
 
 	"github.com/obolnetwork/charon/app/errors"
@@ -31,18 +32,20 @@ import (
 )
 
 // New returns a new fetcher instance.
-func New(eth2Cl eth2wrap.Client) (*Fetcher, error) {
+func New(eth2Cl eth2wrap.Client, feeRecipientAddress string) (*Fetcher, error) {
 	return &Fetcher{
-		eth2Cl: eth2Cl,
+		eth2Cl:              eth2Cl,
+		feeRecipientAddress: feeRecipientAddress,
 	}, nil
 }
 
 // Fetcher fetches proposed duty data.
 type Fetcher struct {
-	eth2Cl           eth2wrap.Client
-	subs             []func(context.Context, core.Duty, core.UnsignedDataSet) error
-	aggSigDBFunc     func(context.Context, core.Duty, core.PubKey) (core.SignedData, error)
-	awaitAttDataFunc func(ctx context.Context, slot int64, commIdx int64) (*eth2p0.AttestationData, error)
+	eth2Cl              eth2wrap.Client
+	feeRecipientAddress string
+	subs                []func(context.Context, core.Duty, core.UnsignedDataSet) error
+	aggSigDBFunc        func(context.Context, core.Duty, core.PubKey) (core.SignedData, error)
+	awaitAttDataFunc    func(ctx context.Context, slot int64, commIdx int64) (*eth2p0.AttestationData, error)
 }
 
 // Subscribe registers a callback for fetched duties.
@@ -224,6 +227,15 @@ func (f *Fetcher) fetchProposerData(ctx context.Context, slot int64, defSet core
 			return nil, err
 		}
 
+		// Ensure fee recipient is correctly populated in block.
+		if block.Version == spec.DataVersionBellatrix {
+			actual := fmt.Sprintf("%#x", block.Bellatrix.Body.ExecutionPayload.FeeRecipient)
+			if actual != f.feeRecipientAddress {
+				log.Warn(ctx, "Proposing block with unexpected fee recipient address", nil,
+					z.Str("expected", f.feeRecipientAddress), z.Str("actual", actual))
+			}
+		}
+
 		coreBlock, err := core.NewVersionedBeaconBlock(block)
 		if err != nil {
 			return nil, errors.Wrap(err, "new block")
@@ -259,6 +271,15 @@ func (f *Fetcher) fetchBuilderProposerData(ctx context.Context, slot int64, defS
 			return nil, err
 		}
 
+		// Ensure fee recipient is correctly populated in block.
+		if block.Version == spec.DataVersionBellatrix {
+			actual := fmt.Sprintf("%#x", block.Bellatrix.Body.ExecutionPayloadHeader.FeeRecipient)
+			if actual != f.feeRecipientAddress {
+				log.Warn(ctx, "Proposing block with unexpected fee recipient address", nil,
+					z.Str("expected", f.feeRecipientAddress), z.Str("actual", actual))
+			}
+		}
+
 		coreBlock, err := core.NewVersionedBlindedBeaconBlock(block)
 		if err != nil {
 			return nil, errors.Wrap(err, "new block")

core/fetcher/fetcher_test.go

@@ -17,6 +17,7 @@ package fetcher_test
 
 import (
 	"context"
+	"fmt"
 	"testing"
 
 	eth2v1 "github.com/attestantio/go-eth2-client/api/v1"
@@ -70,7 +71,7 @@ func TestFetchAttester(t *testing.T) {
 	duty := core.NewAttesterDuty(slot)
 	bmock, err := beaconmock.New()
 	require.NoError(t, err)
-	fetch, err := fetcher.New(bmock)
+	fetch, err := fetcher.New(bmock, "")
 	require.NoError(t, err)
 
 	fetch.Subscribe(func(ctx context.Context, resDuty core.Duty, resDataSet core.UnsignedDataSet) error {
@@ -154,7 +155,7 @@ func TestFetchAggregator(t *testing.T) {
 		return nil, errors.New("expected unknown root")
 	}
 
-	fetch, err := fetcher.New(bmock)
+	fetch, err := fetcher.New(bmock, "")
 	require.NoError(t, err)
 
 	fetch.RegisterAggSigDB(func(ctx context.Context, duty core.Duty, key core.PubKey) (core.SignedData, error) {
@@ -225,9 +226,10 @@ func TestFetchProposer(t *testing.T) {
 	ctx := context.Background()
 
 	const (
-		slot  = 1
-		vIdxA = 2
-		vIdxB = 3
+		slot             = 1
+		vIdxA            = 2
+		vIdxB            = 3
+		feeRecipientAddr = "0x0000000000000000000000000000000000000000"
 	)
 
 	pubkeysByIdx := map[eth2p0.ValidatorIndex]core.PubKey{
@@ -258,7 +260,7 @@ func TestFetchProposer(t *testing.T) {
 
 	bmock, err := beaconmock.New()
 	require.NoError(t, err)
-	fetch, err := fetcher.New(bmock)
+	fetch, err := fetcher.New(bmock, feeRecipientAddr)
 	require.NoError(t, err)
 
 	fetch.RegisterAggSigDB(func(ctx context.Context, duty core.Duty, key core.PubKey) (core.SignedData, error) {
@@ -273,6 +275,7 @@ func TestFetchProposer(t *testing.T) {
 		slotA, err := dutyDataA.Slot()
 		require.NoError(t, err)
 		require.EqualValues(t, slot, slotA)
+		require.Equal(t, feeRecipientAddr, fmt.Sprintf("%#x", dutyDataA.Bellatrix.Body.ExecutionPayload.FeeRecipient))
 		assertRandao(t, randaoByPubKey[pubkeysByIdx[vIdxA]].Signature().ToETH2(), dutyDataA)
 
 		dutyDataB := resDataSet[pubkeysByIdx[vIdxB]].(core.VersionedBeaconBlock)
@@ -292,9 +295,10 @@ func TestFetchBuilderProposer(t *testing.T) {
 	ctx := context.Background()
 
 	const (
-		slot  = 1
-		vIdxA = 2
-		vIdxB = 3
+		slot             = 1
+		vIdxA            = 2
+		vIdxB            = 3
+		feeRecipientAddr = "0x0000000000000000000000000000000000000000"
 	)
 
 	pubkeysByIdx := map[eth2p0.ValidatorIndex]core.PubKey{
@@ -325,7 +329,7 @@ func TestFetchBuilderProposer(t *testing.T) {
 
 	bmock, err := beaconmock.New()
 	require.NoError(t, err)
-	fetch, err := fetcher.New(bmock)
+	fetch, err := fetcher.New(bmock, feeRecipientAddr)
 	require.NoError(t, err)
 
 	fetch.RegisterAggSigDB(func(ctx context.Context, duty core.Duty, key core.PubKey) (core.SignedData, error) {
@@ -340,6 +344,7 @@ func TestFetchBuilderProposer(t *testing.T) {
 		slotA, err := dutyDataA.Slot()
 		require.NoError(t, err)
 		require.EqualValues(t, slot, slotA)
+		require.Equal(t, feeRecipientAddr, fmt.Sprintf("%#x", dutyDataA.Bellatrix.Body.ExecutionPayloadHeader.FeeRecipient))
 		assertRandaoBlindedBlock(t, randaoByPubKey[pubkeysByIdx[vIdxA]].Signature().ToETH2(), dutyDataA)
 
 		dutyDataB := resDataSet[pubkeysByIdx[vIdxB]].(core.VersionedBlindedBeaconBlock)

core/validatorapi/router.go

@@ -169,6 +169,11 @@ func NewRouter(h Handler, eth2Cl eth2wrap.Client) (*mux.Router, error) {
 			Path:    "/eth/v1/beacon/pool/sync_committees",
 			Handler: submitSyncCommitteeMessages(h),
 		},
+		{
+			Name:    "submit_proposal_preparations",
+			Path:    "/eth/v1/validator/prepare_beacon_proposer",
+			Handler: submitProposalPreparations(),
+		},
 	}
 
 	r := mux.NewRouter()
@@ -650,6 +655,14 @@ func submitSyncCommitteeMessages(s eth2client.SyncCommitteeMessagesSubmitter) ha
 	}
 }
 
+// submitProposalPreparations swallows fee-recipient-address from validator client as it should be
+// configured by charon from cluster-lock.json and VC need not be configured with correct fee-recipient-address.
+func submitProposalPreparations() handlerFunc {
+	return func(context.Context, map[string]string, url.Values, []byte) (interface{}, error) {
+		return nil, nil
+	}
+}
+
 // proxyHandler returns a reverse proxy handler.
 func proxyHandler(eth2Cl eth2wrap.Client) http.HandlerFunc {
 	return func(w http.ResponseWriter, r *http.Request) {

testutil/beaconmock/beaconmock.go

@@ -149,6 +149,7 @@ type Mock struct {
 	SubmitSyncCommitteeContributionsFunc     func(ctx context.Context, contributionAndProofs []*altair.SignedContributionAndProof) error
 	SyncCommitteeContributionFunc            func(ctx context.Context, slot eth2p0.Slot, subcommitteeIndex uint64, beaconBlockRoot eth2p0.Root) (*altair.SyncCommitteeContribution, error)
 	SubmitSyncCommitteeSubscriptionsFunc     func(ctx context.Context, subscriptions []*eth2v1.SyncCommitteeSubscription) error
+	SubmitProposalPreparationsFunc           func(ctx context.Context, preparations []*eth2v1.ProposalPreparation) error
 }
 
 func (m Mock) SubmitAttestations(ctx context.Context, attestations []*eth2p0.Attestation) error {
@@ -255,6 +256,10 @@ func (m Mock) SubmitSyncCommitteeSubscriptions(ctx context.Context, subscription
 	return m.SubmitSyncCommitteeSubscriptionsFunc(ctx, subscriptions)
 }
 
+func (m Mock) SubmitProposalPreparations(ctx context.Context, preparations []*eth2v1.ProposalPreparation) error {
+	return m.SubmitProposalPreparationsFunc(ctx, preparations)
+}
+
 func (m Mock) SlotsPerEpoch(ctx context.Context) (uint64, error) {
 	return m.SlotsPerEpochFunc(ctx)
 }

testutil/beaconmock/options.go

@@ -530,6 +530,9 @@ func defaultMock(httpMock HTTPMock, httpServer *http.Server, clock clockwork.Clo
 		SlotsPerEpochFunc: func(ctx context.Context) (uint64, error) {
 			return httpMock.SlotsPerEpoch(ctx)
 		},
+		SubmitProposalPreparationsFunc: func(_ context.Context, _ []*eth2v1.ProposalPreparation) error {
+			return nil
+		},
 	}
 }