Add aggregate attestation strategy.

The "best" aggregate attestation strategy obtains aggregate attestations
from all listed nodes, scores them according to their attestation
coverage, and signs and broadcasts the one with the highest coverage.

The "first" aggregate attestation strategy signs and broadcasts the
first aggregate attestation returned from all listed nodes.

docs/configuration.md

@@ -55,13 +55,24 @@ strategies:
       - localhost:5052
   # The attestationdata strategy obtains attestation data from multiple sources.
   attestationdata:
-    # style can be 'best', which obtains attestations from all nodes and selects the best, or 'first', which uses the first returned
+    # style can be 'best', which obtains attestation data from all nodes and selects the best, or 'first', which uses the first returned
     style: best
     # beacon-node-addresses are the addresses of beacon nodes to use for this strategy.
     beacon-node-addresses:
       - localhost:4000
       - localhost:5051
       - localhost:5052
+  # The aggregateattestation strategy obtains aggregate attestations from multiple sources.
+  # Note that the list of nodes here must be a subset of those in the attestationdata strategy.  If not, the nodes will not have
+  # been gathering the attestations to aggregate and will error when the aggregate request is made.
+  aggregateattestation:
+    # style can be 'best', which obtains aggregates from all nodes and selects the best, or 'first', which uses the first returned
+    style: best
+    # beacon-node-addresses are the addresses of beacon nodes to use for this strategy.
+    # Note that prysm nodes are not supported at current in this strategy.
+    beacon-node-addresses:
+      - localhost:5051
+      - localhost:5052
 ```
 
 ## Logging

main.go

@@ -53,6 +53,8 @@ import (
 	multinodesubmitter "github.com/attestantio/vouch/services/submitter/multinode"
 	"github.com/attestantio/vouch/services/validatorsmanager"
 	standardvalidatorsmanager "github.com/attestantio/vouch/services/validatorsmanager/standard"
+	bestaggregateattestationstrategy "github.com/attestantio/vouch/strategies/aggregateattestation/best"
+	firstaggregateattestationstrategy "github.com/attestantio/vouch/strategies/aggregateattestation/first"
 	bestattestationdatastrategy "github.com/attestantio/vouch/strategies/attestationdata/best"
 	firstattestationdatastrategy "github.com/attestantio/vouch/strategies/attestationdata/first"
 	bestbeaconblockproposalstrategy "github.com/attestantio/vouch/strategies/beaconblockproposal/best"
@@ -342,17 +344,18 @@ func startServices(ctx context.Context, majordomo majordomo.Service) error {
 		return errors.Wrap(err, "failed to start attester service")
 	}
 
-	log.Trace().Msg("Starting beacon attestation aggregator")
-	var aggregationAttester standardattestationaggregator.Parameter
-	if provider, isProvider := eth2Client.(eth2client.AggregateAttestationProvider); isProvider {
-		aggregationAttester = standardattestationaggregator.WithAggregateAttestationDataProvider(provider)
-	} else {
-		aggregationAttester = standardattestationaggregator.WithPrysmAggregateAttestationDataProvider(eth2Client.(eth2client.PrysmAggregateAttestationProvider))
+	log.Trace().Msg("Selecting aggregate attestation provider")
+	aggregateAttestationProvider, prysmAggregateAttestationProvider, err := selectAggregateAttestationProvider(ctx, monitor, eth2Client)
+	if err != nil {
+		return errors.Wrap(err, "failed to select aggregate attestation provider")
 	}
+
+	log.Trace().Msg("Starting beacon attestation aggregator")
 	attestationAggregator, err := standardattestationaggregator.New(ctx,
 		standardattestationaggregator.WithLogLevel(logLevel(viper.GetString("attestationaggregator.log-level"))),
 		standardattestationaggregator.WithTargetAggregatorsPerCommitteeProvider(eth2Client.(eth2client.TargetAggregatorsPerCommitteeProvider)),
-		aggregationAttester,
+		standardattestationaggregator.WithAggregateAttestationProvider(aggregateAttestationProvider),
+		standardattestationaggregator.WithPrysmAggregateAttestationProvider(prysmAggregateAttestationProvider),
 		standardattestationaggregator.WithAggregateAttestationsSubmitter(eth2Client.(eth2client.AggregateAttestationsSubmitter)),
 		standardattestationaggregator.WithMonitor(monitor.(metrics.AttestationAggregationMonitor)),
 		standardattestationaggregator.WithValidatingAccountsProvider(accountManager.(accountmanager.ValidatingAccountsProvider)),
@@ -700,6 +703,75 @@ func selectAttestationDataProvider(ctx context.Context,
 	return attestationDataProvider, nil
 }
 
+func selectAggregateAttestationProvider(ctx context.Context,
+	monitor metrics.Service,
+	eth2Client eth2client.Service,
+) (
+	eth2client.AggregateAttestationProvider,
+	eth2client.PrysmAggregateAttestationProvider,
+	error,
+) {
+	var aggregateAttestationProvider eth2client.AggregateAttestationProvider
+	var prysmAggregateAttestationProvider eth2client.PrysmAggregateAttestationProvider
+	var err error
+	switch viper.GetString("strategies.aggregateattestation.style") {
+	case "best":
+		log.Info().Msg("Starting best aggregate attestation strategy")
+		aggregateAttestationProviders := make(map[string]eth2client.AggregateAttestationProvider)
+		for _, address := range viper.GetStringSlice("strategies.aggregateattestation.beacon-node-addresses") {
+			client, err := fetchClient(ctx, address)
+			if err != nil {
+				return nil, nil, errors.Wrap(err, fmt.Sprintf("failed to fetch client %s for aggregate attestation strategy", address))
+			}
+			if _, isProvider := client.(eth2client.PrysmAggregateAttestationProvider); isProvider {
+				log.Warn().Str("provider", address).Msg("Cannot use prysm in aggregate attestation strategy; ignoring provider")
+				continue
+			}
+			aggregateAttestationProviders[address] = client.(eth2client.AggregateAttestationProvider)
+		}
+		aggregateAttestationProvider, err = bestaggregateattestationstrategy.New(ctx,
+			bestaggregateattestationstrategy.WithClientMonitor(monitor.(metrics.ClientMonitor)),
+			bestaggregateattestationstrategy.WithProcessConcurrency(viper.GetInt64("process-concurrency")),
+			bestaggregateattestationstrategy.WithLogLevel(logLevel(viper.GetString("strategies.aggregateattestation.log-level"))),
+			bestaggregateattestationstrategy.WithAggregateAttestationProviders(aggregateAttestationProviders),
+		)
+		if err != nil {
+			return nil, nil, errors.Wrap(err, "failed to start best aggregate attestation strategy")
+		}
+	case "first":
+		log.Info().Msg("Starting first aggregate attestation strategy")
+		aggregateAttestationProviders := make(map[string]eth2client.AggregateAttestationProvider)
+		for _, address := range viper.GetStringSlice("strategies.aggregateattestation.beacon-node-addresses") {
+			client, err := fetchClient(ctx, address)
+			if err != nil {
+				return nil, nil, errors.Wrap(err, fmt.Sprintf("failed to fetch client %s for aggregate attestation strategy", address))
+			}
+			if _, isProvider := client.(eth2client.PrysmAggregateAttestationProvider); isProvider {
+				log.Warn().Str("provider", address).Msg("Cannot use prysm in aggregate attestation strategy; ignoring provider")
+				continue
+			}
+			aggregateAttestationProviders[address] = client.(eth2client.AggregateAttestationProvider)
+		}
+		aggregateAttestationProvider, err = firstaggregateattestationstrategy.New(ctx,
+			firstaggregateattestationstrategy.WithClientMonitor(monitor.(metrics.ClientMonitor)),
+			firstaggregateattestationstrategy.WithLogLevel(logLevel(viper.GetString("strategies.aggregateattestation.log-level"))),
+			firstaggregateattestationstrategy.WithAggregateAttestationProviders(aggregateAttestationProviders),
+		)
+		if err != nil {
+			return nil, nil, errors.Wrap(err, "failed to start first aggregate attestation strategy")
+		}
+	default:
+		log.Info().Msg("Starting simple aggregate attestation strategy")
+		if _, isProvider := eth2Client.(eth2client.AggregateAttestationProvider); isProvider {
+			aggregateAttestationProvider = eth2Client.(eth2client.AggregateAttestationProvider)
+		} else {
+			prysmAggregateAttestationProvider = eth2Client.(eth2client.PrysmAggregateAttestationProvider)
+		}
+	}
+
+	return aggregateAttestationProvider, prysmAggregateAttestationProvider, nil
+}
+
 func selectBeaconBlockProposalProvider(ctx context.Context,
 	monitor metrics.Service,
 	eth2Client eth2client.Service,

mock/eth2client.go

@@ -400,6 +400,104 @@ func (m *SleepyAttestationDataProvider) AttestationData(ctx context.Context, slo
 	return m.next.AttestationData(ctx, slot, committeeIndex)
 }
 
+// AggregateAttestationProvider is a mock for eth2client.AggregateAttestationProvider.
+type AggregateAttestationProvider struct{}
+
+// NewAggregateAttestationProvider returns a mock attestation data provider.
+func NewAggregateAttestationProvider() eth2client.AggregateAttestationProvider {
+	return &AggregateAttestationProvider{}
+}
+
+// AggregateAttestation is a mock.
+func (m *AggregateAttestationProvider) AggregateAttestation(ctx context.Context, slot spec.Slot, attestationDataRoot spec.Root) (*spec.Attestation, error) {
+	aggregationBits := bitfield.NewBitlist(128)
+	aggregationBits.SetBitAt(1, true)
+	aggregationBits.SetBitAt(3, true)
+	aggregationBits.SetBitAt(8, true)
+	aggregationBits.SetBitAt(12, true)
+	aggregationBits.SetBitAt(65, true)
+	aggregationBits.SetBitAt(77, true)
+	return &spec.Attestation{
+		AggregationBits: aggregationBits,
+		Data: &spec.AttestationData{
+			Slot:  slot,
+			Index: 1,
+			BeaconBlockRoot: spec.Root([32]byte{
+				0x00, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0a, 0x0b, 0x0c, 0x0d, 0x0e, 0x0f,
+				0x10, 0x11, 0x12, 0x13, 0x14, 0x15, 0x16, 0x17, 0x18, 0x19, 0x1a, 0x1b, 0x1c, 0x1d, 0x1e, 0x1f,
+			}),
+			Source: &spec.Checkpoint{
+				Epoch: 1,
+				Root: spec.Root([32]byte{
+					0x20, 0x21, 0x22, 0x23, 0x24, 0x25, 0x26, 0x27, 0x28, 0x29, 0x2a, 0x2b, 0x2c, 0x2d, 0x2e, 0x2f,
+					0x30, 0x31, 0x32, 0x33, 0x34, 0x35, 0x36, 0x37, 0x38, 0x39, 0x3a, 0x3b, 0x3c, 0x3d, 0x3e, 0x3f,
+				}),
+			},
+			Target: &spec.Checkpoint{
+				Epoch: 2,
+				Root: spec.Root([32]byte{
+					0x40, 0x41, 0x42, 0x43, 0x44, 0x45, 0x46, 0x47, 0x48, 0x49, 0x4a, 0x4b, 0x4c, 0x4d, 0x4e, 0x4f,
+					0x30, 0x31, 0x32, 0x33, 0x34, 0x35, 0x36, 0x37, 0x38, 0x39, 0x3a, 0x3b, 0x3c, 0x3d, 0x3e, 0x3f,
+				}),
+			},
+		},
+		Signature: spec.BLSSignature([96]byte{
+			0x00, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0a, 0x0b, 0x0c, 0x0d, 0x0e, 0x0f,
+			0x10, 0x11, 0x12, 0x13, 0x14, 0x15, 0x16, 0x17, 0x18, 0x19, 0x1a, 0x1b, 0x1c, 0x1d, 0x1e, 0x1f,
+			0x20, 0x21, 0x22, 0x23, 0x24, 0x25, 0x26, 0x27, 0x28, 0x29, 0x2a, 0x2b, 0x2c, 0x2d, 0x2e, 0x2f,
+			0x30, 0x31, 0x32, 0x33, 0x34, 0x35, 0x36, 0x37, 0x38, 0x39, 0x3a, 0x3b, 0x3c, 0x3d, 0x3e, 0x3f,
+			0x40, 0x41, 0x42, 0x43, 0x44, 0x45, 0x46, 0x47, 0x48, 0x49, 0x4a, 0x4b, 0x4c, 0x4d, 0x4e, 0x4f,
+			0x50, 0x51, 0x52, 0x53, 0x54, 0x55, 0x56, 0x57, 0x58, 0x59, 0x5a, 0x5b, 0x5c, 0x5d, 0x5e, 0x5f,
+		}),
+	}, nil
+}
+
+// ErroringAggregateAttestationProvider is a mock for eth2client.AggregateAttestationProvider.
+type ErroringAggregateAttestationProvider struct{}
+
+// NewErroringAggregateAttestationProvider returns a mock attestation data provider.
+func NewErroringAggregateAttestationProvider() eth2client.AggregateAttestationProvider {
+	return &ErroringAggregateAttestationProvider{}
+}
+
+// AggregateAttestation is a mock.
+func (m *ErroringAggregateAttestationProvider) AggregateAttestation(ctx context.Context, slot spec.Slot, attestationDataRoot spec.Root) (*spec.Attestation, error) {
+	return nil, errors.New("mock error")
+}
+
+// NilAggregateAttestationProvider is a mock for eth2client.AggregateAttestationProvider.
+type NilAggregateAttestationProvider struct{}
+
+// NewNilAggregateAttestationProvider returns a mock attestation data provider.
+func NewNilAggregateAttestationProvider() eth2client.AggregateAttestationProvider {
+	return &NilAggregateAttestationProvider{}
+}
+
+// AggregateAttestation is a mock.
+func (m *NilAggregateAttestationProvider) AggregateAttestation(ctx context.Context, slot spec.Slot, attestationDataRoot spec.Root) (*spec.Attestation, error) {
+	return nil, nil
+}
+
+// SleepyAggregateAttestationProvider is a mock for eth2client.AggregateAttestationProvider.
+type SleepyAggregateAttestationProvider struct {
+	wait time.Duration
+	next eth2client.AggregateAttestationProvider
+}
+
+// NewSleepyAggregateAttestationProvider returns a mock attestation data provider.
+func NewSleepyAggregateAttestationProvider(wait time.Duration, next eth2client.AggregateAttestationProvider) eth2client.AggregateAttestationProvider {
+	return &SleepyAggregateAttestationProvider{
+		wait: wait,
+		next: next,
+	}
+}
+
+// AggregateAttestation is a mock.
+func (m *SleepyAggregateAttestationProvider) AggregateAttestation(ctx context.Context, slot spec.Slot, attestationDataRoot spec.Root) (*spec.Attestation, error) {
+	time.Sleep(m.wait)
+	return m.next.AggregateAttestation(ctx, slot, attestationDataRoot)
+}
+
 // BeaconProposerDomainProvider is a mock for eth2client.BeaconProposerDomainProvider.
 type BeaconProposerDomainProvider struct{}
 

services/attestationaggregator/standard/parameters.go

@@ -82,15 +82,15 @@ func WithValidatingAccountsProvider(provider accountmanager.ValidatingAccountsPr
 	})
 }
 
-// WithAggregateAttestationDataProvider sets the aggregate attestation provider.
-func WithAggregateAttestationDataProvider(provider eth2client.AggregateAttestationProvider) Parameter {
+// WithAggregateAttestationProvider sets the aggregate attestation provider.
+func WithAggregateAttestationProvider(provider eth2client.AggregateAttestationProvider) Parameter {
 	return parameterFunc(func(p *parameters) {
 		p.aggregateAttestationProvider = provider
 	})
 }
 
-// WithPrysmAggregateAttestationDataProvider sets the non-spec aggregate attestation provider.
-func WithPrysmAggregateAttestationDataProvider(provider eth2client.PrysmAggregateAttestationProvider) Parameter {
+// WithPrysmAggregateAttestationProvider sets the non-spec aggregate attestation provider.
+func WithPrysmAggregateAttestationProvider(provider eth2client.PrysmAggregateAttestationProvider) Parameter {
 	return parameterFunc(func(p *parameters) {
 		p.prysmAggregateAttestationProvider = provider
 	})

strategies/aggregateattestation/best/aggregateattestation.go

@@ -0,0 +1,95 @@
+// Copyright © 2020 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 best
+
+import (
+	"context"
+	"time"
+
+	eth2client "github.com/attestantio/go-eth2-client"
+	spec "github.com/attestantio/go-eth2-client/spec/phase0"
+	"github.com/pkg/errors"
+)
+
+type aggregateAttestationResponse struct {
+	aggregate *spec.Attestation
+	score     float64
+}
+
+// AggregateAttestation provides the aggregate attestation from a number of beacon nodes.
+func (s *Service) AggregateAttestation(ctx context.Context, slot spec.Slot, attestationDataRoot spec.Root) (*spec.Attestation, error) {
+	started := time.Now()
+
+	// We create a cancelable context with a timeout.  If the context times out we take the best to date.
+	ctx, cancel := context.WithTimeout(ctx, s.timeout)
+
+	respCh := make(chan *aggregateAttestationResponse, len(s.aggregateAttestationProviders))
+	errCh := make(chan error, len(s.aggregateAttestationProviders))
+	// Kick off the requests.
+	for name, provider := range s.aggregateAttestationProviders {
+		go func(ctx context.Context,
+			name string,
+			provider eth2client.AggregateAttestationProvider,
+			respCh chan *aggregateAttestationResponse,
+			errCh chan error,
+		) {
+			aggregate, err := provider.AggregateAttestation(ctx, slot, attestationDataRoot)
+			s.clientMonitor.ClientOperation(name, "aggregate attestation", err == nil, time.Since(started))
+			if err != nil {
+				errCh <- err
+				return
+			}
+			log.Trace().Str("provider", name).Dur("elapsed", time.Since(started)).Msg("Obtained aggregate attestation")
+			if aggregate == nil {
+				return
+			}
+
+			score := s.scoreAggregateAttestation(ctx, name, aggregate)
+			respCh <- &aggregateAttestationResponse{
+				aggregate: aggregate,
+				score:     score,
+			}
+		}(ctx, name, provider, respCh, errCh)
+	}
+
+	// Wait for all responses (or context done).
+	responded := 0
+	errored := 0
+	bestScore := float64(0)
+	var bestAggregateAttestation *spec.Attestation
+	for responded+errored != len(s.aggregateAttestationProviders) {
+		select {
+		case <-ctx.Done():
+			// Anyone not responded by now is considered errored.
+			errored = len(s.aggregateAttestationProviders) - responded
+			log.Info().Dur("elapsed", time.Since(started)).Int("responded", responded).Int("errored", errored).Msg("Timed out waiting for responses")
+		case <-errCh:
+			errored++
+		case resp := <-respCh:
+			responded++
+			if bestAggregateAttestation == nil || resp.score > bestScore {
+				bestAggregateAttestation = resp.aggregate
+				bestScore = resp.score
+			}
+		}
+	}
+	cancel()
+
+	if bestAggregateAttestation == nil {
+		return nil, errors.New("no aggregate attestations received")
+	}
+	log.Trace().Stringer("aggregate_attestation", bestAggregateAttestation).Float64("score", bestScore).Msg("Selected best aggregate attestation")
+
+	return bestAggregateAttestation, nil
+}

strategies/aggregateattestation/best/aggregateattestation_internal_test.go

@@ -0,0 +1,86 @@
+// Copyright © 2020 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 best
+
+import (
+	"context"
+	"testing"
+
+	eth2client "github.com/attestantio/go-eth2-client"
+	spec "github.com/attestantio/go-eth2-client/spec/phase0"
+	"github.com/attestantio/vouch/mock"
+	"github.com/prysmaticlabs/go-bitfield"
+	"github.com/rs/zerolog"
+	"github.com/stretchr/testify/require"
+)
+
+// populatedBitlist creates a populated bitlist.
+func populatedBitlist(size uint64, set uint64) bitfield.Bitlist {
+	res := bitfield.NewBitlist(size)
+	for i := uint64(0); i < set; i++ {
+		res.SetBitAt(i, true)
+	}
+
+	return res
+}
+
+func TestScore(t *testing.T) {
+	ctx := context.Background()
+
+	s, err := New(ctx,
+		WithLogLevel(zerolog.Disabled),
+		WithAggregateAttestationProviders(map[string]eth2client.AggregateAttestationProvider{
+			"good": mock.NewAggregateAttestationProvider(),
+		}),
+	)
+	require.NoError(t, err)
+
+	tests := []struct {
+		name      string
+		aggregate *spec.Attestation
+		score     float64
+	}{
+		{
+			name:  "Nil",
+			score: 0,
+		},
+		{
+			name: "Empty",
+			aggregate: &spec.Attestation{
+				AggregationBits: populatedBitlist(100, 0),
+				Data: &spec.AttestationData{
+					Slot: 5,
+				},
+			},
+			score: 0,
+		},
+		{
+			name: "Full",
+			aggregate: &spec.Attestation{
+				AggregationBits: populatedBitlist(100, 100),
+				Data: &spec.AttestationData{
+					Slot: 5,
+				},
+			},
+			score: 1,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			score := s.scoreAggregateAttestation(ctx, "test", test.aggregate)
+			require.Equal(t, test.score, score)
+		})
+	}
+}

strategies/aggregateattestation/best/aggregateattestation_test.go

@@ -0,0 +1,115 @@
+// Copyright © 2020 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 best_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	eth2client "github.com/attestantio/go-eth2-client"
+	spec "github.com/attestantio/go-eth2-client/spec/phase0"
+	"github.com/attestantio/vouch/mock"
+	"github.com/attestantio/vouch/strategies/aggregateattestation/best"
+	"github.com/rs/zerolog"
+	"github.com/stretchr/testify/require"
+)
+
+func TestAggregateAttestation(t *testing.T) {
+	tests := []struct {
+		name                string
+		params              []best.Parameter
+		slot                spec.Slot
+		attestationDataRoot spec.Root
+		err                 string
+	}{
+		{
+			name: "Good",
+			params: []best.Parameter{
+				best.WithLogLevel(zerolog.Disabled),
+				best.WithAggregateAttestationProviders(map[string]eth2client.AggregateAttestationProvider{
+					"good": mock.NewAggregateAttestationProvider(),
+				}),
+			},
+			slot: 12345,
+			attestationDataRoot: spec.Root{
+				0x00, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0a, 0x0b, 0x0c, 0x0d, 0x0e, 0x0f,
+				0x10, 0x11, 0x12, 0x13, 0x14, 0x15, 0x16, 0x17, 0x18, 0x19, 0x1a, 0x1b, 0x1c, 0x1d, 0x1e, 0x1f,
+			},
+		},
+		{
+			name: "Timeout",
+			params: []best.Parameter{
+				best.WithLogLevel(zerolog.Disabled),
+				best.WithTimeout(time.Second),
+				best.WithAggregateAttestationProviders(map[string]eth2client.AggregateAttestationProvider{
+					"sleepy": mock.NewSleepyAggregateAttestationProvider(5*time.Second, mock.NewAggregateAttestationProvider()),
+				}),
+			},
+			slot: 12345,
+			attestationDataRoot: spec.Root{
+				0x00, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0a, 0x0b, 0x0c, 0x0d, 0x0e, 0x0f,
+				0x10, 0x11, 0x12, 0x13, 0x14, 0x15, 0x16, 0x17, 0x18, 0x19, 0x1a, 0x1b, 0x1c, 0x1d, 0x1e, 0x1f,
+			},
+			err: "no aggregate attestations received",
+		},
+		{
+			name: "NilResponse",
+			params: []best.Parameter{
+				best.WithLogLevel(zerolog.Disabled),
+				best.WithTimeout(time.Second),
+				best.WithAggregateAttestationProviders(map[string]eth2client.AggregateAttestationProvider{
+					"nil": mock.NewNilAggregateAttestationProvider(),
+				}),
+			},
+			slot: 12345,
+			attestationDataRoot: spec.Root{
+				0x00, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0a, 0x0b, 0x0c, 0x0d, 0x0e, 0x0f,
+				0x10, 0x11, 0x12, 0x13, 0x14, 0x15, 0x16, 0x17, 0x18, 0x19, 0x1a, 0x1b, 0x1c, 0x1d, 0x1e, 0x1f,
+			},
+			err: "no aggregate attestations received",
+		},
+		{
+			name: "GoodMixed",
+			params: []best.Parameter{
+				best.WithLogLevel(zerolog.Disabled),
+				best.WithTimeout(2 * time.Second),
+				best.WithAggregateAttestationProviders(map[string]eth2client.AggregateAttestationProvider{
+					"error":  mock.NewErroringAggregateAttestationProvider(),
+					"sleepy": mock.NewSleepyAggregateAttestationProvider(time.Second, mock.NewAggregateAttestationProvider()),
+				}),
+			},
+			slot: 12345,
+			attestationDataRoot: spec.Root{
+				0x00, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0a, 0x0b, 0x0c, 0x0d, 0x0e, 0x0f,
+				0x10, 0x11, 0x12, 0x13, 0x14, 0x15, 0x16, 0x17, 0x18, 0x19, 0x1a, 0x1b, 0x1c, 0x1d, 0x1e, 0x1f,
+			},
+		},
+	}
+
+	for _, test := range tests {
+		s, err := best.New(context.Background(), test.params...)
+		require.NoError(t, err)
+
+		t.Run(test.name, func(t *testing.T) {
+			aggregate, err := s.AggregateAttestation(context.Background(), test.slot, test.attestationDataRoot)
+			if test.err != "" {
+				require.EqualError(t, err, test.err)
+			} else {
+				require.NoError(t, err)
+				require.NotNil(t, aggregate)
+			}
+		})
+	}
+}

strategies/aggregateattestation/best/parameters.go

@@ -0,0 +1,112 @@
+// Copyright © 2020 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 best is a strategy that obtains attestation from multiple
+// nodes and selects the best one.
+package best
+
+import (
+	"context"
+	"runtime"
+	"time"
+
+	eth2client "github.com/attestantio/go-eth2-client"
+	"github.com/attestantio/vouch/services/metrics"
+	nullmetrics "github.com/attestantio/vouch/services/metrics/null"
+	"github.com/pkg/errors"
+	"github.com/rs/zerolog"
+)
+
+type parameters struct {
+	logLevel                      zerolog.Level
+	clientMonitor                 metrics.ClientMonitor
+	processConcurrency            int64
+	aggregateAttestationProviders map[string]eth2client.AggregateAttestationProvider
+	timeout                       time.Duration
+}
+
+// Parameter is the interface for service parameters.
+type Parameter interface {
+	apply(*parameters)
+}
+
+type parameterFunc func(*parameters)
+
+func (f parameterFunc) apply(p *parameters) {
+	f(p)
+}
+
+// WithLogLevel sets the log level for the module.
+func WithLogLevel(logLevel zerolog.Level) Parameter {
+	return parameterFunc(func(p *parameters) {
+		p.logLevel = logLevel
+	})
+}
+
+// WithClientMonitor sets the client monitor for the service.
+func WithClientMonitor(monitor metrics.ClientMonitor) Parameter {
+	return parameterFunc(func(p *parameters) {
+		p.clientMonitor = monitor
+	})
+}
+
+// WithProcessConcurrency sets the concurrency for the service.
+func WithProcessConcurrency(concurrency int64) Parameter {
+	return parameterFunc(func(p *parameters) {
+		p.processConcurrency = concurrency
+	})
+}
+
+// WithAggregateAttestationProviders sets the aggregate attestation providers.
+func WithAggregateAttestationProviders(providers map[string]eth2client.AggregateAttestationProvider) Parameter {
+	return parameterFunc(func(p *parameters) {
+		p.aggregateAttestationProviders = providers
+	})
+}
+
+// WithTimeout sets the timeout for beacon block proposal requests.
+func WithTimeout(timeout time.Duration) Parameter {
+	return parameterFunc(func(p *parameters) {
+		p.timeout = timeout
+	})
+}
+
+// parseAndCheckParameters parses and checks parameters to ensure that mandatory parameters are present and correct.
+func parseAndCheckParameters(params ...Parameter) (*parameters, error) {
+	parameters := parameters{
+		logLevel:           zerolog.GlobalLevel(),
+		timeout:            2 * time.Second,
+		clientMonitor:      nullmetrics.New(context.Background()),
+		processConcurrency: int64(runtime.GOMAXPROCS(-1)),
+	}
+	for _, p := range params {
+		if params != nil {
+			p.apply(&parameters)
+		}
+	}
+
+	if parameters.timeout == 0 {
+		return nil, errors.New("no timeout specified")
+	}
+	if parameters.clientMonitor == nil {
+		return nil, errors.New("no client monitor specified")
+	}
+	if parameters.processConcurrency == 0 {
+		return nil, errors.New("no process concurrency specified")
+	}
+	if len(parameters.aggregateAttestationProviders) == 0 {
+		return nil, errors.New("no aggregate attestation providers specified")
+	}
+
+	return &parameters, nil
+}

strategies/aggregateattestation/best/score.go

@@ -0,0 +1,47 @@
+// Copyright © 2020 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 best
+
+import (
+	"context"
+
+	spec "github.com/attestantio/go-eth2-client/spec/phase0"
+)
+
+// scoreAggregateAttestation generates a score for an aggregate attestation.
+// The score is relative to the completeness of the aggregate.
+func (s *Service) scoreAggregateAttestation(ctx context.Context,
+	name string,
+	aggregate *spec.Attestation,
+) float64 {
+	if aggregate == nil {
+		return 0
+	}
+
+	included := 0
+	total := aggregate.AggregationBits.Len()
+	for i := uint64(0); i < total; i++ {
+		if aggregate.AggregationBits.BitAt(i) {
+			included++
+		}
+	}
+	score := float64(included) / float64(total)
+
+	log.Trace().
+		Str("provider", name).
+		Uint64("attestation_slot", uint64(aggregate.Data.Slot)).
+		Float64("score", score).
+		Msg("Scored aggregate attestation")
+	return score
+}

strategies/aggregateattestation/best/service.go

@@ -0,0 +1,59 @@
+// Copyright © 2020 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 best
+
+import (
+	"context"
+	"time"
+
+	eth2client "github.com/attestantio/go-eth2-client"
+	"github.com/attestantio/vouch/services/metrics"
+	"github.com/pkg/errors"
+	"github.com/rs/zerolog"
+	zerologger "github.com/rs/zerolog/log"
+)
+
+// Service is the provider for attestation data.
+type Service struct {
+	clientMonitor                 metrics.ClientMonitor
+	processConcurrency            int64
+	aggregateAttestationProviders map[string]eth2client.AggregateAttestationProvider
+	timeout                       time.Duration
+}
+
+// module-wide log.
+var log zerolog.Logger
+
+// New creates a new attestation data strategy.
+func New(ctx context.Context, params ...Parameter) (*Service, error) {
+	parameters, err := parseAndCheckParameters(params...)
+	if err != nil {
+		return nil, errors.Wrap(err, "problem with parameters")
+	}
+
+	// Set logging.
+	log = zerologger.With().Str("strategy", "aggregateattestation").Str("impl", "best").Logger()
+	if parameters.logLevel != log.GetLevel() {
+		log = log.Level(parameters.logLevel)
+	}
+
+	s := &Service{
+		timeout:                       parameters.timeout,
+		clientMonitor:                 parameters.clientMonitor,
+		processConcurrency:            parameters.processConcurrency,
+		aggregateAttestationProviders: parameters.aggregateAttestationProviders,
+	}
+
+	return s, nil
+}

strategies/aggregateattestation/best/service_test.go

@@ -0,0 +1,114 @@
+// Copyright © 2020 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 best_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	eth2client "github.com/attestantio/go-eth2-client"
+	"github.com/attestantio/vouch/mock"
+	"github.com/attestantio/vouch/strategies/aggregateattestation/best"
+	"github.com/rs/zerolog"
+	"github.com/stretchr/testify/require"
+)
+
+func TestService(t *testing.T) {
+	aggregateAttestationProviders := map[string]eth2client.AggregateAttestationProvider{
+		"localhost:1": mock.NewAggregateAttestationProvider(),
+	}
+
+	tests := []struct {
+		name   string
+		params []best.Parameter
+		err    string
+	}{
+		{
+			name: "TimeoutZero",
+			params: []best.Parameter{
+				best.WithLogLevel(zerolog.TraceLevel),
+				best.WithTimeout(0),
+				best.WithAggregateAttestationProviders(aggregateAttestationProviders),
+			},
+			err: "problem with parameters: no timeout specified",
+		},
+		{
+			name: "ClientMonitorMissing",
+			params: []best.Parameter{
+				best.WithLogLevel(zerolog.TraceLevel),
+				best.WithClientMonitor(nil),
+				best.WithAggregateAttestationProviders(aggregateAttestationProviders),
+			},
+			err: "problem with parameters: no client monitor specified",
+		},
+		{
+			name: "AggregateAttestationProvidersNil",
+			params: []best.Parameter{
+				best.WithLogLevel(zerolog.TraceLevel),
+				best.WithAggregateAttestationProviders(nil),
+			},
+			err: "problem with parameters: no aggregate attestation providers specified",
+		},
+		{
+			name: "ProcessConcurrencyZero",
+			params: []best.Parameter{
+				best.WithLogLevel(zerolog.TraceLevel),
+				best.WithAggregateAttestationProviders(aggregateAttestationProviders),
+				best.WithProcessConcurrency(0),
+			},
+			err: "problem with parameters: no process concurrency specified",
+		},
+		{
+			name: "AggregateAttestationProvidersEmpty",
+			params: []best.Parameter{
+				best.WithLogLevel(zerolog.TraceLevel),
+				best.WithAggregateAttestationProviders(map[string]eth2client.AggregateAttestationProvider{}),
+			},
+			err: "problem with parameters: no aggregate attestation providers specified",
+		},
+		{
+			name: "Good",
+			params: []best.Parameter{
+				best.WithLogLevel(zerolog.TraceLevel),
+				best.WithTimeout(10 * time.Second),
+				best.WithAggregateAttestationProviders(aggregateAttestationProviders),
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			_, err := best.New(context.Background(), test.params...)
+			if test.err != "" {
+				require.EqualError(t, err, test.err)
+			} else {
+				require.NoError(t, err)
+			}
+		})
+	}
+}
+
+func TestInterfaces(t *testing.T) {
+	aggregateAttestationProviders := map[string]eth2client.AggregateAttestationProvider{
+		"localhost:1": mock.NewAggregateAttestationProvider(),
+	}
+
+	s, err := best.New(context.Background(),
+		best.WithLogLevel(zerolog.Disabled),
+		best.WithAggregateAttestationProviders(aggregateAttestationProviders),
+	)
+	require.NoError(t, err)
+	require.Implements(t, (*eth2client.AggregateAttestationProvider)(nil), s)
+}

strategies/aggregateattestation/first/aggregateattestation.go

@@ -0,0 +1,65 @@
+// Copyright © 2020 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 first
+
+import (
+	"context"
+	"time"
+
+	eth2client "github.com/attestantio/go-eth2-client"
+	spec "github.com/attestantio/go-eth2-client/spec/phase0"
+	"github.com/pkg/errors"
+)
+
+// AggregateAttestation provides the aggregate attestation from a number of beacon nodes.
+func (s *Service) AggregateAttestation(ctx context.Context, slot spec.Slot, attestationDataRoot spec.Root) (*spec.Attestation, error) {
+	started := time.Now()
+
+	// We create a cancelable context with a timeout.  When a provider responds we cancel the context to cancel the other requests.
+	ctx, cancel := context.WithTimeout(ctx, s.timeout)
+
+	respCh := make(chan *spec.Attestation, 1)
+	for name, provider := range s.aggregateAttestationProviders {
+		go func(ctx context.Context,
+			name string,
+			provider eth2client.AggregateAttestationProvider,
+			ch chan *spec.Attestation) {
+			log := log.With().Str("provider", name).Uint64("slot", uint64(slot)).Logger()
+
+			aggregate, err := provider.AggregateAttestation(ctx, slot, attestationDataRoot)
+			s.clientMonitor.ClientOperation(name, "aggregate attestation", err == nil, time.Since(started))
+			if err != nil {
+				log.Warn().Dur("elapsed", time.Since(started)).Err(err).Msg("Failed to obtain aggregate attestation")
+				return
+			}
+			if aggregate == nil {
+				log.Warn().Dur("elapsed", time.Since(started)).Err(err).Msg("Returned empty aggregate attestation")
+				return
+			}
+			log.Trace().Str("provider", name).Dur("elapsed", time.Since(started)).Msg("Obtained aggregate attestation")
+
+			ch <- aggregate
+		}(ctx, name, provider, respCh)
+	}
+
+	select {
+	case <-ctx.Done():
+		cancel()
+		log.Warn().Msg("Failed to obtain aggregate attestation before timeout")
+		return nil, errors.New("failed to obtain aggregate attestation before timeout")
+	case aggregate := <-respCh:
+		cancel()
+		return aggregate, nil
+	}
+}

strategies/aggregateattestation/first/aggregateattestation_test.go

@@ -0,0 +1,116 @@
+// Copyright © 2020 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 first_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	eth2client "github.com/attestantio/go-eth2-client"
+	spec "github.com/attestantio/go-eth2-client/spec/phase0"
+	"github.com/attestantio/vouch/mock"
+	"github.com/attestantio/vouch/strategies/aggregateattestation/first"
+	"github.com/rs/zerolog"
+	"github.com/stretchr/testify/require"
+)
+
+func TestAggregateAttestation(t *testing.T) {
+	tests := []struct {
+		name                string
+		params              []first.Parameter
+		slot                spec.Slot
+		attestationDataRoot spec.Root
+		err                 string
+	}{
+		{
+			name: "Good",
+			params: []first.Parameter{
+				first.WithLogLevel(zerolog.Disabled),
+				first.WithAggregateAttestationProviders(map[string]eth2client.AggregateAttestationProvider{
+					"good": mock.NewAggregateAttestationProvider(),
+				}),
+			},
+			slot: 12345,
+			attestationDataRoot: spec.Root{
+				0x00, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0a, 0x0b, 0x0c, 0x0d, 0x0e, 0x0f,
+				0x10, 0x11, 0x12, 0x13, 0x14, 0x15, 0x16, 0x17, 0x18, 0x19, 0x1a, 0x1b, 0x1c, 0x1d, 0x1e, 0x1f,
+			},
+		},
+		{
+			name: "Timeout",
+			params: []first.Parameter{
+				first.WithLogLevel(zerolog.Disabled),
+				first.WithTimeout(time.Second),
+				first.WithAggregateAttestationProviders(map[string]eth2client.AggregateAttestationProvider{
+					"sleepy": mock.NewSleepyAggregateAttestationProvider(5*time.Second, mock.NewAggregateAttestationProvider()),
+				}),
+			},
+			slot: 12345,
+			attestationDataRoot: spec.Root{
+				0x00, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0a, 0x0b, 0x0c, 0x0d, 0x0e, 0x0f,
+				0x10, 0x11, 0x12, 0x13, 0x14, 0x15, 0x16, 0x17, 0x18, 0x19, 0x1a, 0x1b, 0x1c, 0x1d, 0x1e, 0x1f,
+			},
+			err: "failed to obtain aggregate attestation before timeout",
+		},
+		{
+			name: "NilResponse",
+			params: []first.Parameter{
+				first.WithLogLevel(zerolog.Disabled),
+				first.WithTimeout(time.Second),
+				first.WithAggregateAttestationProviders(map[string]eth2client.AggregateAttestationProvider{
+					"nil": mock.NewNilAggregateAttestationProvider(),
+				}),
+			},
+			slot: 12345,
+			attestationDataRoot: spec.Root{
+				0x00, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0a, 0x0b, 0x0c, 0x0d, 0x0e, 0x0f,
+				0x10, 0x11, 0x12, 0x13, 0x14, 0x15, 0x16, 0x17, 0x18, 0x19, 0x1a, 0x1b, 0x1c, 0x1d, 0x1e, 0x1f,
+			},
+			// Nil response is invalid, so expect a timeout.
+			err: "failed to obtain aggregate attestation before timeout",
+		},
+		{
+			name: "GoodMixed",
+			params: []first.Parameter{
+				first.WithLogLevel(zerolog.Disabled),
+				first.WithTimeout(2 * time.Second),
+				first.WithAggregateAttestationProviders(map[string]eth2client.AggregateAttestationProvider{
+					"error":  mock.NewErroringAggregateAttestationProvider(),
+					"sleepy": mock.NewSleepyAggregateAttestationProvider(time.Second, mock.NewAggregateAttestationProvider()),
+				}),
+			},
+			slot: 12345,
+			attestationDataRoot: spec.Root{
+				0x00, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0a, 0x0b, 0x0c, 0x0d, 0x0e, 0x0f,
+				0x10, 0x11, 0x12, 0x13, 0x14, 0x15, 0x16, 0x17, 0x18, 0x19, 0x1a, 0x1b, 0x1c, 0x1d, 0x1e, 0x1f,
+			},
+		},
+	}
+
+	for _, test := range tests {
+		s, err := first.New(context.Background(), test.params...)
+		require.NoError(t, err)
+
+		t.Run(test.name, func(t *testing.T) {
+			attestationData, err := s.AggregateAttestation(context.Background(), test.slot, test.attestationDataRoot)
+			if test.err != "" {
+				require.EqualError(t, err, test.err)
+			} else {
+				require.NoError(t, err)
+				require.NotNil(t, attestationData)
+			}
+		})
+	}
+}

strategies/aggregateattestation/first/parameters.go

@@ -0,0 +1,99 @@
+// Copyright © 2020 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 first is a strategy that obtains attestation from multiple
+// nodes and selects the first one returned.
+package first
+
+import (
+	"context"
+	"time"
+
+	eth2client "github.com/attestantio/go-eth2-client"
+	"github.com/attestantio/vouch/services/metrics"
+	nullmetrics "github.com/attestantio/vouch/services/metrics/null"
+	"github.com/pkg/errors"
+	"github.com/rs/zerolog"
+)
+
+type parameters struct {
+	logLevel                      zerolog.Level
+	clientMonitor                 metrics.ClientMonitor
+	aggregateAttestationProviders map[string]eth2client.AggregateAttestationProvider
+	timeout                       time.Duration
+}
+
+// Parameter is the interface for service parameters.
+type Parameter interface {
+	apply(*parameters)
+}
+
+type parameterFunc func(*parameters)
+
+func (f parameterFunc) apply(p *parameters) {
+	f(p)
+}
+
+// WithLogLevel sets the log level for the module.
+func WithLogLevel(logLevel zerolog.Level) Parameter {
+	return parameterFunc(func(p *parameters) {
+		p.logLevel = logLevel
+	})
+}
+
+// WithClientMonitor sets the client monitor for the service.
+func WithClientMonitor(monitor metrics.ClientMonitor) Parameter {
+	return parameterFunc(func(p *parameters) {
+		p.clientMonitor = monitor
+	})
+}
+
+// WithAggregateAttestationProviders sets the aggregate attestation providers.
+func WithAggregateAttestationProviders(providers map[string]eth2client.AggregateAttestationProvider) Parameter {
+	return parameterFunc(func(p *parameters) {
+		p.aggregateAttestationProviders = providers
+	})
+}
+
+// WithTimeout sets the timeout for beacon block proposal requests.
+func WithTimeout(timeout time.Duration) Parameter {
+	return parameterFunc(func(p *parameters) {
+		p.timeout = timeout
+	})
+}
+
+// parseAndCheckParameters parses and checks parameters to ensure that mandatory parameters are present and correct.
+func parseAndCheckParameters(params ...Parameter) (*parameters, error) {
+	parameters := parameters{
+		logLevel:      zerolog.GlobalLevel(),
+		timeout:       2 * time.Second,
+		clientMonitor: nullmetrics.New(context.Background()),
+	}
+	for _, p := range params {
+		if params != nil {
+			p.apply(&parameters)
+		}
+	}
+
+	if parameters.timeout == 0 {
+		return nil, errors.New("no timeout specified")
+	}
+	if parameters.clientMonitor == nil {
+		return nil, errors.New("no client monitor specified")
+	}
+	if len(parameters.aggregateAttestationProviders) == 0 {
+		return nil, errors.New("no aggregate attestation providers specified")
+	}
+
+	return &parameters, nil
+}

strategies/aggregateattestation/first/service.go

@@ -0,0 +1,57 @@
+// Copyright © 2020 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 first
+
+import (
+	"context"
+	"time"
+
+	eth2client "github.com/attestantio/go-eth2-client"
+	"github.com/attestantio/vouch/services/metrics"
+	"github.com/pkg/errors"
+	"github.com/rs/zerolog"
+	zerologger "github.com/rs/zerolog/log"
+)
+
+// Service is the provider for attestation data.
+type Service struct {
+	clientMonitor                 metrics.ClientMonitor
+	aggregateAttestationProviders map[string]eth2client.AggregateAttestationProvider
+	timeout                       time.Duration
+}
+
+// module-wide log.
+var log zerolog.Logger
+
+// New creates a new attestation data strategy.
+func New(ctx context.Context, params ...Parameter) (*Service, error) {
+	parameters, err := parseAndCheckParameters(params...)
+	if err != nil {
+		return nil, errors.Wrap(err, "problem with parameters")
+	}
+
+	// Set logging.
+	log = zerologger.With().Str("strategy", "aggregateattestation").Str("impl", "first").Logger()
+	if parameters.logLevel != log.GetLevel() {
+		log = log.Level(parameters.logLevel)
+	}
+
+	s := &Service{
+		aggregateAttestationProviders: parameters.aggregateAttestationProviders,
+		timeout:                       parameters.timeout,
+		clientMonitor:                 parameters.clientMonitor,
+	}
+
+	return s, nil
+}

strategies/aggregateattestation/first/service_test.go

@@ -0,0 +1,105 @@
+// Copyright © 2020 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 first_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	eth2client "github.com/attestantio/go-eth2-client"
+	"github.com/attestantio/vouch/mock"
+	"github.com/attestantio/vouch/strategies/aggregateattestation/first"
+	"github.com/rs/zerolog"
+	"github.com/stretchr/testify/require"
+)
+
+func TestService(t *testing.T) {
+	aggregateAttestationProviders := map[string]eth2client.AggregateAttestationProvider{
+		"localhost:1": mock.NewAggregateAttestationProvider(),
+	}
+
+	tests := []struct {
+		name   string
+		params []first.Parameter
+		err    string
+	}{
+		{
+			name: "TimeoutZero",
+			params: []first.Parameter{
+				first.WithLogLevel(zerolog.TraceLevel),
+				first.WithTimeout(0),
+				first.WithAggregateAttestationProviders(aggregateAttestationProviders),
+			},
+			err: "problem with parameters: no timeout specified",
+		},
+		{
+			name: "ClientMonitorMissing",
+			params: []first.Parameter{
+				first.WithLogLevel(zerolog.TraceLevel),
+				first.WithClientMonitor(nil),
+				first.WithAggregateAttestationProviders(aggregateAttestationProviders),
+			},
+			err: "problem with parameters: no client monitor specified",
+		},
+		{
+			name: "AggregateAttestationProvidersNil",
+			params: []first.Parameter{
+				first.WithLogLevel(zerolog.TraceLevel),
+				first.WithAggregateAttestationProviders(nil),
+			},
+			err: "problem with parameters: no aggregate attestation providers specified",
+		},
+		{
+			name: "AggregateAttestationProvidersEmpty",
+			params: []first.Parameter{
+				first.WithLogLevel(zerolog.TraceLevel),
+				first.WithAggregateAttestationProviders(map[string]eth2client.AggregateAttestationProvider{}),
+			},
+			err: "problem with parameters: no aggregate attestation providers specified",
+		},
+		{
+			name: "Good",
+			params: []first.Parameter{
+				first.WithLogLevel(zerolog.TraceLevel),
+				first.WithTimeout(10 * time.Second),
+				first.WithAggregateAttestationProviders(aggregateAttestationProviders),
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			_, err := first.New(context.Background(), test.params...)
+			if test.err != "" {
+				require.EqualError(t, err, test.err)
+			} else {
+				require.NoError(t, err)
+			}
+		})
+	}
+}
+
+func TestInterfaces(t *testing.T) {
+	aggregateAttestationProviders := map[string]eth2client.AggregateAttestationProvider{
+		"localhost:1": mock.NewAggregateAttestationProvider(),
+	}
+
+	s, err := first.New(context.Background(),
+		first.WithLogLevel(zerolog.Disabled),
+		first.WithAggregateAttestationProviders(aggregateAttestationProviders),
+	)
+	require.NoError(t, err)
+	require.Implements(t, (*eth2client.AggregateAttestationProvider)(nil), s)
+}