core/fetcher: add DutyProposer to fetcher (#305)

Adds DutyProposer in Fetcher

category: feature
ticket: #222

app/app.go

@@ -250,6 +250,8 @@ func wireCoreWorkflow(ctx context.Context, life *lifecycle.Manager, conf Config,
 		opts := []beaconmock.Option{
 			beaconmock.WithSlotDuration(time.Second),
 			beaconmock.WithDeterministicDuties(100),
+			// TODO(dhruv): remove this when DutyProposer is in place
+			beaconmock.WithNoProposerDuties(),
 			beaconmock.WithValidatorSet(createMockValidators(pubkeys)),
 		}
 		opts = append(opts, conf.TestConfig.SimnetBMockOpts...)

core/encode.go

@@ -18,6 +18,7 @@ import (
 	"encoding/json"
 
 	eth2v1 "github.com/attestantio/go-eth2-client/api/v1"
+	"github.com/attestantio/go-eth2-client/spec"
 	eth2p0 "github.com/attestantio/go-eth2-client/spec/phase0"
 
 	"github.com/obolnetwork/charon/app/errors"
@@ -161,3 +162,24 @@ func EncodeRandaoAggSignedData(randao eth2p0.BLSSignature) AggSignedData {
 func DecodeRandaoAggSignedData(data AggSignedData) eth2p0.BLSSignature {
 	return data.Signature.ToETH2()
 }
+
+// EncodeProposerUnsignedData returns the proposer data as an encoded UnsignedData.
+func EncodeProposerUnsignedData(proData *spec.VersionedBeaconBlock) (UnsignedData, error) {
+	b, err := json.Marshal(proData)
+	if err != nil {
+		return nil, errors.Wrap(err, "marshal proposer data")
+	}
+
+	return b, nil
+}
+
+// DecodeProposerUnsignedData returns the proposer data as an encoded UnsignedData.
+func DecodeProposerUnsignedData(unsignedData UnsignedData) (*spec.VersionedBeaconBlock, error) {
+	proData := new(spec.VersionedBeaconBlock)
+	err := json.Unmarshal(unsignedData, proData)
+	if err != nil {
+		return nil, errors.Wrap(err, "unmarshal proposer data")
+	}
+
+	return proData, nil
+}

core/encode_test.go

@@ -17,6 +17,7 @@ package core_test
 import (
 	"testing"
 
+	"github.com/attestantio/go-eth2-client/spec"
 	"github.com/stretchr/testify/require"
 
 	"github.com/obolnetwork/charon/core"
@@ -127,3 +128,22 @@ func TestEncodeProposerFetchArg(t *testing.T) {
 	require.Equal(t, arg1, arg2)
 	require.Equal(t, proDuty1, proDuty2)
 }
+
+func TestEncodeProposerUnsignedData(t *testing.T) {
+	proData1 := &spec.VersionedBeaconBlock{
+		Version: spec.DataVersionPhase0,
+		Phase0:  testutil.RandomBeaconBlock(),
+	}
+
+	data1, err := core.EncodeProposerUnsignedData(proData1)
+	require.NoError(t, err)
+
+	proData2, err := core.DecodeProposerUnsignedData(data1)
+	require.NoError(t, err)
+
+	data2, err := core.EncodeProposerUnsignedData(proData2)
+	require.NoError(t, err)
+
+	require.Equal(t, proData1, proData2)
+	require.Equal(t, data1, data2)
+}

core/fetcher/fetcher.go

@@ -28,6 +28,7 @@ import (
 // eth2Provider defines the eth2 provider subset used by this package.
 type eth2Provider interface {
 	eth2client.AttestationDataProvider
+	eth2client.BeaconBlockProposalProvider
 }
 
 // New returns a new fetcher instance.
@@ -44,8 +45,9 @@ func New(eth2Svc eth2client.Service) (*Fetcher, error) {
 
 // Fetcher fetches proposed duty data.
 type Fetcher struct {
-	eth2Cl eth2Provider
-	subs   []func(context.Context, core.Duty, core.UnsignedDataSet) error
+	eth2Cl       eth2Provider
+	subs         []func(context.Context, core.Duty, core.UnsignedDataSet) error
+	aggSigDBFunc func(context.Context, core.Duty, core.PubKey) (core.AggSignedData, error)
 }
 
 // Subscribe registers a callback for fetched duties.
@@ -63,8 +65,10 @@ func (f *Fetcher) Fetch(ctx context.Context, duty core.Duty, argSet core.FetchAr
 
 	switch duty.Type {
 	case core.DutyProposer:
-		// TODO(dhruv): Add support for proposer here
-		return nil
+		unsignedSet, err = f.fetchProposerData(ctx, duty.Slot, argSet)
+		if err != nil {
+			return errors.Wrap(err, "fetch proposer data")
+		}
 	case core.DutyAttester:
 		unsignedSet, err = f.fetchAttesterData(ctx, duty.Slot, argSet)
 		if err != nil {
@@ -84,6 +88,12 @@ func (f *Fetcher) Fetch(ctx context.Context, duty core.Duty, argSet core.FetchAr
 	return nil
 }
 
+// RegisterAggSigDB registers a function to get resolved aggregated signed data from the AggSigDB.
+// Note: This is not thread safe should be called *before* Fetch.
+func (f *Fetcher) RegisterAggSigDB(fn func(context.Context, core.Duty, core.PubKey) (core.AggSignedData, error)) {
+	f.aggSigDBFunc = fn
+}
+
 // fetchAttesterData returns the fetched attestation data set for committees and validators in the arg set.
 func (f *Fetcher) fetchAttesterData(ctx context.Context, slot int64, argSet core.FetchArgSet,
 ) (core.UnsignedDataSet, error) {
@@ -122,3 +132,36 @@ func (f *Fetcher) fetchAttesterData(ctx context.Context, slot int64, argSet core
 
 	return resp, nil
 }
+
+func (f *Fetcher) fetchProposerData(ctx context.Context, slot int64, argSet core.FetchArgSet) (core.UnsignedDataSet, error) {
+	resp := make(core.UnsignedDataSet)
+	for pubkey := range argSet {
+		// Fetch previously aggregated randao reveal from AggSigDB
+		dutyRandao := core.Duty{
+			Slot: slot,
+			Type: core.DutyRandao,
+		}
+		randao, err := f.aggSigDBFunc(ctx, dutyRandao, pubkey)
+		if err != nil {
+			return nil, err
+		}
+		randaoEth2 := core.DecodeRandaoAggSignedData(randao)
+
+		// TODO(dhruv): what to do with graffiti?
+		// passing empty graffiti since it is not required in API
+		var graffiti [32]byte
+		block, err := f.eth2Cl.BeaconBlockProposal(ctx, eth2p0.Slot(uint64(slot)), randaoEth2, graffiti[:])
+		if err != nil {
+			return nil, err
+		}
+
+		dutyData, err := core.EncodeProposerUnsignedData(block)
+		if err != nil {
+			return nil, errors.Wrap(err, "encode proposer data")
+		}
+
+		resp[pubkey] = dutyData
+	}
+
+	return resp, nil
+}

core/fetcher/fetcher_test.go

@@ -98,3 +98,82 @@ func TestFetchAttester(t *testing.T) {
 	err = fetch.Fetch(ctx, duty, argSet)
 	require.NoError(t, err)
 }
+
+func TestFetchProposer(t *testing.T) {
+	ctx := context.Background()
+
+	const (
+		slot  = 1
+		vIdxA = 2
+		vIdxB = 3
+	)
+
+	pubkeysByIdx := map[eth2p0.ValidatorIndex]core.PubKey{
+		vIdxA: testutil.RandomCorePubKey(t),
+		vIdxB: testutil.RandomCorePubKey(t),
+	}
+
+	dutyA := eth2v1.ProposerDuty{
+		Slot:           slot,
+		ValidatorIndex: vIdxA,
+	}
+	fetchArgA, err := core.EncodeProposerFetchArg(&dutyA)
+	require.NoError(t, err)
+
+	dutyB := eth2v1.ProposerDuty{
+		Slot:           slot,
+		ValidatorIndex: vIdxB,
+	}
+	fetchArgB, err := core.EncodeProposerFetchArg(&dutyB)
+	require.NoError(t, err)
+
+	argSet := core.FetchArgSet{
+		pubkeysByIdx[vIdxA]: fetchArgA,
+		pubkeysByIdx[vIdxB]: fetchArgB,
+	}
+	duty := core.Duty{Type: core.DutyProposer, Slot: slot}
+
+	randaoA := core.AggSignedData{
+		Data:      nil,
+		Signature: testutil.RandomCoreSignature(),
+	}
+	randaoB := core.AggSignedData{
+		Data:      nil,
+		Signature: testutil.RandomCoreSignature(),
+	}
+	randaoByPubKey := map[core.PubKey]core.AggSignedData{
+		pubkeysByIdx[vIdxA]: randaoA,
+		pubkeysByIdx[vIdxB]: randaoB,
+	}
+
+	bmock, err := beaconmock.New()
+	require.NoError(t, err)
+	fetch, err := fetcher.New(bmock)
+	require.NoError(t, err)
+
+	fetch.RegisterAggSigDB(func(ctx context.Context, duty core.Duty, key core.PubKey) (core.AggSignedData, error) {
+		return randaoByPubKey[key], nil
+	})
+
+	fetch.Subscribe(func(ctx context.Context, resDuty core.Duty, resDataSet core.UnsignedDataSet) error {
+		require.Equal(t, duty, resDuty)
+		require.Len(t, resDataSet, 2)
+
+		dataA := resDataSet[pubkeysByIdx[vIdxA]]
+		dutyDataA, err := core.DecodeProposerUnsignedData(dataA)
+		require.NoError(t, err)
+		require.EqualValues(t, slot, dutyDataA.Phase0.Slot)
+		require.EqualValues(t, randaoByPubKey[pubkeysByIdx[vIdxA]].Signature.ToETH2(), dutyDataA.Phase0.Body.RANDAOReveal)
+
+		dataB := resDataSet[pubkeysByIdx[vIdxB]]
+		dutyDataB, err := core.DecodeProposerUnsignedData(dataB)
+		require.NoError(t, err)
+		require.EqualValues(t, slot, dutyDataB.Phase0.Slot)
+		require.EqualValues(t, randaoByPubKey[pubkeysByIdx[vIdxB]].Signature.ToETH2(), dutyDataB.Phase0.Body.RANDAOReveal)
+
+		return nil
+	})
+
+	err = fetch.Fetch(ctx, duty, argSet)
+	require.NoError(t, err)
+}

core/interfaces.go

@@ -33,6 +33,10 @@ type Fetcher interface {
 
 	// Subscribe registers a callback for proposed unsigned duty data sets.
 	Subscribe(func(context.Context, Duty, UnsignedDataSet) error)
+
+	// RegisterAggSigDB registers a function to get resolved aggregated
+	// signed data from the AggSigDB (e.g., randao reveals).
+	RegisterAggSigDB(func(context.Context, Duty, PubKey) (AggSignedData, error))
 }
 
 // DutyDB persists unsigned duty data sets and makes it available for querying. It also acts

docs/architecture.md

@@ -214,7 +214,7 @@ type Fetcher interface {
   // Subscribe registers a callback for proposed duty data sets.
   Subscribe(func(context.Context, Duty, UnsignedDataSet) error)
 
-  // RegisterAggDB registers a function to resolved aggregated
+  // RegisterAggSigDB registers a function to resolved aggregated
   // signed data from the AggSigDB (e.g., randao reveals).
   RegisterAggSigDB(func(context.Context, Duty, PubKey) (AggSignedData, error))
 }

testutil/beaconmock/beaconmock.go

@@ -43,6 +43,7 @@ import (
 
 	eth2client "github.com/attestantio/go-eth2-client"
 	eth2v1 "github.com/attestantio/go-eth2-client/api/v1"
+	"github.com/attestantio/go-eth2-client/spec"
 	eth2p0 "github.com/attestantio/go-eth2-client/spec/phase0"
 	"github.com/jonboulle/clockwork"
 
@@ -51,13 +52,14 @@ import (
 
 // Interface assertions.
 var (
-	_ HTTPMock                           = (*Mock)(nil)
-	_ eth2client.AttestationDataProvider = (*Mock)(nil)
-	_ eth2client.AttestationsSubmitter   = (*Mock)(nil)
-	_ eth2client.AttesterDutiesProvider  = (*Mock)(nil)
-	_ eth2client.ProposerDutiesProvider  = (*Mock)(nil)
-	_ eth2client.Service                 = (*Mock)(nil)
-	_ eth2client.ValidatorsProvider      = (*Mock)(nil)
+	_ HTTPMock                               = (*Mock)(nil)
+	_ eth2client.AttestationDataProvider     = (*Mock)(nil)
+	_ eth2client.AttestationsSubmitter       = (*Mock)(nil)
+	_ eth2client.AttesterDutiesProvider      = (*Mock)(nil)
+	_ eth2client.BeaconBlockProposalProvider = (*Mock)(nil)
+	_ eth2client.ProposerDutiesProvider      = (*Mock)(nil)
+	_ eth2client.Service                     = (*Mock)(nil)
+	_ eth2client.ValidatorsProvider          = (*Mock)(nil)
 )
 
 // New returns a new beacon client mock configured with the default and provided options.
@@ -120,14 +122,15 @@ type Mock struct {
 	overrides  []staticOverride
 	clock      clockwork.Clock
 
-	AttestationDataFunc    func(context.Context, eth2p0.Slot, eth2p0.CommitteeIndex) (*eth2p0.AttestationData, error)
-	AttesterDutiesFunc     func(context.Context, eth2p0.Epoch, []eth2p0.ValidatorIndex) ([]*eth2v1.AttesterDuty, error)
-	ProposerDutiesFunc     func(context.Context, eth2p0.Epoch, []eth2p0.ValidatorIndex) ([]*eth2v1.ProposerDuty, error)
-	SubmitAttestationsFunc func(context.Context, []*eth2p0.Attestation) error
-	ValidatorsByPubKeyFunc func(context.Context, string, []eth2p0.BLSPubKey) (map[eth2p0.ValidatorIndex]*eth2v1.Validator, error)
-	ValidatorsFunc         func(context.Context, string, []eth2p0.ValidatorIndex) (map[eth2p0.ValidatorIndex]*eth2v1.Validator, error)
-	GenesisTimeFunc        func(ctx context.Context) (time.Time, error)
-	NodeSyncingFunc        func(context.Context) (*eth2v1.SyncState, error)
+	AttestationDataFunc     func(context.Context, eth2p0.Slot, eth2p0.CommitteeIndex) (*eth2p0.AttestationData, error)
+	AttesterDutiesFunc      func(context.Context, eth2p0.Epoch, []eth2p0.ValidatorIndex) ([]*eth2v1.AttesterDuty, error)
+	BeaconBlockProposalFunc func(ctx context.Context, slot eth2p0.Slot, randaoReveal eth2p0.BLSSignature, graffiti []byte) (*spec.VersionedBeaconBlock, error)
+	ProposerDutiesFunc      func(context.Context, eth2p0.Epoch, []eth2p0.ValidatorIndex) ([]*eth2v1.ProposerDuty, error)
+	SubmitAttestationsFunc  func(context.Context, []*eth2p0.Attestation) error
+	ValidatorsByPubKeyFunc  func(context.Context, string, []eth2p0.BLSPubKey) (map[eth2p0.ValidatorIndex]*eth2v1.Validator, error)
+	ValidatorsFunc          func(context.Context, string, []eth2p0.ValidatorIndex) (map[eth2p0.ValidatorIndex]*eth2v1.Validator, error)
+	GenesisTimeFunc         func(ctx context.Context) (time.Time, error)
+	NodeSyncingFunc         func(context.Context) (*eth2v1.SyncState, error)
 }
 
 func (m Mock) SubmitAttestations(ctx context.Context, attestations []*eth2p0.Attestation) error {
@@ -138,6 +141,10 @@ func (m Mock) AttestationData(ctx context.Context, slot eth2p0.Slot, committeeIn
 	return m.AttestationDataFunc(ctx, slot, committeeIndex)
 }
 
+func (m Mock) BeaconBlockProposal(ctx context.Context, slot eth2p0.Slot, randaoReveal eth2p0.BLSSignature, graffiti []byte) (*spec.VersionedBeaconBlock, error) {
+	return m.BeaconBlockProposalFunc(ctx, slot, randaoReveal, graffiti)
+}
+
 func (m Mock) ProposerDuties(ctx context.Context, epoch eth2p0.Epoch, validatorIndices []eth2p0.ValidatorIndex) ([]*eth2v1.ProposerDuty, error) {
 	return m.ProposerDutiesFunc(ctx, epoch, validatorIndices)
 }

testutil/beaconmock/options.go

@@ -26,13 +26,15 @@ import (
 	"time"
 
 	eth2v1 "github.com/attestantio/go-eth2-client/api/v1"
+	"github.com/attestantio/go-eth2-client/spec"
 	eth2p0 "github.com/attestantio/go-eth2-client/spec/phase0"
 	"github.com/jonboulle/clockwork"
 
 	"github.com/obolnetwork/charon/app/errors"
 	"github.com/obolnetwork/charon/app/log"
 	"github.com/obolnetwork/charon/app/z"
 	"github.com/obolnetwork/charon/core"
+	"github.com/obolnetwork/charon/testutil"
 )
 
 // Option defines a functional option to configure the mock beacon client.
@@ -288,6 +290,15 @@ func WithDeterministicDuties(factor int) Option {
 	}
 }
 
+// WithNoProposerDuties configures the mock to override ProposerDutiesFunc to return nothing.
+func WithNoProposerDuties() Option {
+	return func(mock *Mock) {
+		mock.ProposerDutiesFunc = func(ctx context.Context, epoch eth2p0.Epoch, indices []eth2p0.ValidatorIndex) ([]*eth2v1.ProposerDuty, error) {
+			return nil, nil
+		}
+	}
+}
+
 // WithClock configures the mock with the provided clock.
 func WithClock(clock clockwork.Clock) Option {
 	return func(mock *Mock) {
@@ -301,6 +312,28 @@ func defaultMock(httpMock HTTPMock, httpServer *http.Server, clock clockwork.Clo
 		clock:      clock,
 		HTTPMock:   httpMock,
 		httpServer: httpServer,
+		BeaconBlockProposalFunc: func(ctx context.Context, slot eth2p0.Slot, randaoReveal eth2p0.BLSSignature, graffiti []byte) (*spec.VersionedBeaconBlock, error) {
+			return &spec.VersionedBeaconBlock{
+				Version: spec.DataVersionPhase0,
+				Phase0: &eth2p0.BeaconBlock{
+					Slot: slot,
+					Body: &eth2p0.BeaconBlockBody{
+						RANDAOReveal: randaoReveal,
+						ETH1Data: &eth2p0.ETH1Data{
+							DepositRoot:  testutil.RandomRoot(),
+							DepositCount: 0,
+							BlockHash:    testutil.RandomBytes(),
+						},
+						Graffiti:          testutil.RandomBytes(),
+						ProposerSlashings: []*eth2p0.ProposerSlashing{},
+						AttesterSlashings: []*eth2p0.AttesterSlashing{},
+						Attestations:      []*eth2p0.Attestation{testutil.RandomAttestation(), testutil.RandomAttestation()},
+						Deposits:          []*eth2p0.Deposit{},
+						VoluntaryExits:    []*eth2p0.SignedVoluntaryExit{},
+					},
+				},
+			}, nil
+		},
 		ProposerDutiesFunc: func(ctx context.Context, epoch eth2p0.Epoch, indices []eth2p0.ValidatorIndex) ([]*eth2v1.ProposerDuty, error) {
 			return []*eth2v1.ProposerDuty{}, nil
 		},