2020-09-27 23:46:00 -07:00
|
|
|
// 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 standard
|
|
|
|
|
|
|
|
import (
|
|
|
|
"context"
|
|
|
|
"sync"
|
|
|
|
"time"
|
|
|
|
|
|
|
|
eth2client "github.com/attestantio/go-eth2-client"
|
2020-10-28 08:09:51 -07:00
|
|
|
api "github.com/attestantio/go-eth2-client/api/v1"
|
2021-07-17 23:34:43 -07:00
|
|
|
"github.com/attestantio/go-eth2-client/spec/phase0"
|
2020-09-27 23:46:00 -07:00
|
|
|
"github.com/attestantio/vouch/services/attestationaggregator"
|
|
|
|
"github.com/attestantio/vouch/services/attester"
|
|
|
|
"github.com/attestantio/vouch/services/beaconcommitteesubscriber"
|
|
|
|
"github.com/attestantio/vouch/services/metrics"
|
|
|
|
"github.com/attestantio/vouch/services/submitter"
|
|
|
|
"github.com/pkg/errors"
|
|
|
|
"github.com/rs/zerolog"
|
|
|
|
zerologger "github.com/rs/zerolog/log"
|
|
|
|
"github.com/sasha-s/go-deadlock"
|
2020-11-24 16:02:13 -08:00
|
|
|
e2wtypes "github.com/wealdtech/go-eth2-wallet-types/v2"
|
2020-09-27 23:46:00 -07:00
|
|
|
"golang.org/x/sync/semaphore"
|
|
|
|
)
|
|
|
|
|
|
|
|
// Service is an beacon committee subscriber.
|
|
|
|
type Service struct {
|
|
|
|
monitor metrics.BeaconCommitteeSubscriptionMonitor
|
|
|
|
processConcurrency int64
|
|
|
|
attesterDutiesProvider eth2client.AttesterDutiesProvider
|
|
|
|
attestationAggregator attestationaggregator.Service
|
|
|
|
submitter submitter.BeaconCommitteeSubscriptionsSubmitter
|
|
|
|
}
|
|
|
|
|
|
|
|
// module-wide log.
|
|
|
|
var log zerolog.Logger
|
|
|
|
|
|
|
|
// New creates a new beacon committee subscriber.
|
|
|
|
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("service", "beaconcommitteesubscriber").Str("impl", "standard").Logger()
|
|
|
|
if parameters.logLevel != log.GetLevel() {
|
|
|
|
log = log.Level(parameters.logLevel)
|
|
|
|
}
|
|
|
|
|
|
|
|
s := &Service{
|
|
|
|
processConcurrency: parameters.processConcurrency,
|
|
|
|
monitor: parameters.monitor,
|
|
|
|
attesterDutiesProvider: parameters.attesterDutiesProvider,
|
|
|
|
attestationAggregator: parameters.attestationAggregator,
|
|
|
|
submitter: parameters.beaconCommitteeSubmitter,
|
|
|
|
}
|
|
|
|
|
|
|
|
return s, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
// Subscribe subscribes to beacon committees for a given epoch.
|
|
|
|
// This returns data about the subnets to which we are subscribing.
|
2020-11-14 00:36:49 -08:00
|
|
|
func (s *Service) Subscribe(ctx context.Context,
|
2021-07-17 23:34:43 -07:00
|
|
|
epoch phase0.Epoch,
|
|
|
|
accounts map[phase0.ValidatorIndex]e2wtypes.Account,
|
|
|
|
) (map[phase0.Slot]map[phase0.CommitteeIndex]*beaconcommitteesubscriber.Subscription, error) {
|
2020-11-27 07:26:22 -08:00
|
|
|
if len(accounts) == 0 {
|
|
|
|
// Nothing to do.
|
2021-07-17 23:34:43 -07:00
|
|
|
return map[phase0.Slot]map[phase0.CommitteeIndex]*beaconcommitteesubscriber.Subscription{}, nil
|
2020-11-27 07:26:22 -08:00
|
|
|
}
|
2020-09-27 23:46:00 -07:00
|
|
|
|
2020-11-27 07:26:22 -08:00
|
|
|
started := time.Now()
|
2020-11-14 00:36:49 -08:00
|
|
|
log := log.With().Uint64("epoch", uint64(epoch)).Logger()
|
2020-09-27 23:46:00 -07:00
|
|
|
log.Trace().Msg("Subscribing")
|
|
|
|
|
2021-07-17 23:34:43 -07:00
|
|
|
validatorIndices := make([]phase0.ValidatorIndex, 0, len(accounts))
|
2020-11-24 16:02:13 -08:00
|
|
|
for index := range accounts {
|
|
|
|
validatorIndices = append(validatorIndices, index)
|
2020-09-27 23:46:00 -07:00
|
|
|
}
|
2020-11-24 16:02:13 -08:00
|
|
|
attesterDuties, err := s.attesterDutiesProvider.AttesterDuties(ctx, epoch, validatorIndices)
|
2020-09-27 23:46:00 -07:00
|
|
|
if err != nil {
|
|
|
|
s.monitor.BeaconCommitteeSubscriptionCompleted(started, "failed")
|
|
|
|
return nil, errors.Wrap(err, "failed to obtain attester duties")
|
|
|
|
}
|
2020-11-24 16:02:13 -08:00
|
|
|
|
|
|
|
log.Trace().Dur("elapsed", time.Since(started)).Int("accounts", len(validatorIndices)).Msg("Fetched attester duties")
|
2020-09-27 23:46:00 -07:00
|
|
|
duties, err := attester.MergeDuties(ctx, attesterDuties)
|
|
|
|
if err != nil {
|
|
|
|
s.monitor.BeaconCommitteeSubscriptionCompleted(started, "failed")
|
|
|
|
return nil, errors.Wrap(err, "failed to merge attester duties")
|
|
|
|
}
|
|
|
|
|
|
|
|
subscriptionInfo, err := s.calculateSubscriptionInfo(ctx, epoch, accounts, duties)
|
|
|
|
if err != nil {
|
|
|
|
s.monitor.BeaconCommitteeSubscriptionCompleted(started, "failed")
|
|
|
|
return nil, errors.Wrap(err, "failed to calculate subscription duties")
|
|
|
|
}
|
|
|
|
log.Trace().Dur("elapsed", time.Since(started)).Msg("Calculated subscription info")
|
|
|
|
|
|
|
|
// Update metrics.
|
|
|
|
subscriptions := 0
|
|
|
|
aggregators := 0
|
|
|
|
for _, v := range subscriptionInfo {
|
|
|
|
for _, v2 := range v {
|
|
|
|
subscriptions++
|
2020-10-28 08:09:51 -07:00
|
|
|
if v2.IsAggregator {
|
2020-09-27 23:46:00 -07:00
|
|
|
aggregators++
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
s.monitor.BeaconCommitteeSubscribers(subscriptions)
|
|
|
|
s.monitor.BeaconCommitteeAggregators(aggregators)
|
|
|
|
|
|
|
|
// Submit the subscription information.
|
|
|
|
go func() {
|
|
|
|
log.Trace().Msg("Submitting subscription")
|
2020-10-28 08:09:51 -07:00
|
|
|
subscriptions := make([]*api.BeaconCommitteeSubscription, 0, len(duties))
|
2020-09-27 23:46:00 -07:00
|
|
|
for slot, slotInfo := range subscriptionInfo {
|
|
|
|
for committeeIndex, info := range slotInfo {
|
2020-10-28 08:09:51 -07:00
|
|
|
subscriptions = append(subscriptions, &api.BeaconCommitteeSubscription{
|
|
|
|
ValidatorIndex: info.Duty.ValidatorIndex,
|
|
|
|
Slot: slot,
|
|
|
|
CommitteeIndex: committeeIndex,
|
|
|
|
CommitteesAtSlot: info.Duty.CommitteesAtSlot,
|
|
|
|
IsAggregator: info.IsAggregator,
|
2020-09-27 23:46:00 -07:00
|
|
|
})
|
|
|
|
}
|
|
|
|
}
|
|
|
|
if err := s.submitter.SubmitBeaconCommitteeSubscriptions(ctx, subscriptions); err != nil {
|
|
|
|
log.Error().Err(err).Msg("Failed to submit beacon committees")
|
|
|
|
s.monitor.BeaconCommitteeSubscriptionCompleted(started, "failed")
|
|
|
|
return
|
|
|
|
}
|
|
|
|
log.Trace().Dur("elapsed", time.Since(started)).Msg("Submitted subscription request")
|
|
|
|
s.monitor.BeaconCommitteeSubscriptionCompleted(started, "succeeded")
|
|
|
|
}()
|
|
|
|
|
|
|
|
// Return the subscription info so the calling function knows the subnets to which we are subscribing.
|
|
|
|
return subscriptionInfo, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
// calculateSubscriptionInfo calculates our beacon block attesation subnet requirements given a set of duties.
|
|
|
|
// It returns a map of slot => committee => subscription information.
|
|
|
|
func (s *Service) calculateSubscriptionInfo(ctx context.Context,
|
2021-07-17 23:34:43 -07:00
|
|
|
epoch phase0.Epoch,
|
|
|
|
accounts map[phase0.ValidatorIndex]e2wtypes.Account,
|
2020-09-27 23:46:00 -07:00
|
|
|
duties []*attester.Duty,
|
2021-07-17 23:34:43 -07:00
|
|
|
) (map[phase0.Slot]map[phase0.CommitteeIndex]*beaconcommitteesubscriber.Subscription, error) {
|
2020-09-27 23:46:00 -07:00
|
|
|
|
|
|
|
// Map is slot => committee => info.
|
2021-07-17 23:34:43 -07:00
|
|
|
subscriptionInfo := make(map[phase0.Slot]map[phase0.CommitteeIndex]*beaconcommitteesubscriber.Subscription)
|
2020-09-27 23:46:00 -07:00
|
|
|
subscriptionInfoMutex := deadlock.RWMutex{}
|
|
|
|
|
2020-11-24 16:02:13 -08:00
|
|
|
// // Map is validator ID => account.
|
2021-07-17 23:34:43 -07:00
|
|
|
// accountMap := make(map[phase0.ValidatorIndex]accountmanager.ValidatingAccount, len(accounts))
|
2020-11-24 16:02:13 -08:00
|
|
|
// for _, account := range accounts {
|
|
|
|
// index, err := account.Index(ctx)
|
|
|
|
// if err != nil {
|
|
|
|
// log.Warn().Err(err).Msg("Failed to obtain account index for account map")
|
|
|
|
// continue
|
|
|
|
// }
|
|
|
|
// accountMap[index] = account
|
|
|
|
// }
|
2020-09-27 23:46:00 -07:00
|
|
|
|
|
|
|
// Gather aggregators info in parallel.
|
|
|
|
// Note that it is possible for two validators to be aggregating for the same (slot,committee index) tuple, however
|
|
|
|
// once we have a validator aggregating for a tuple we ignore subsequent validators with the same tuple.
|
|
|
|
sem := semaphore.NewWeighted(s.processConcurrency)
|
|
|
|
var wg sync.WaitGroup
|
|
|
|
for _, duty := range duties {
|
|
|
|
wg.Add(1)
|
|
|
|
go func(ctx context.Context, sem *semaphore.Weighted, wg *sync.WaitGroup, duty *attester.Duty) {
|
|
|
|
defer wg.Done()
|
|
|
|
for i := range duty.ValidatorIndices() {
|
|
|
|
wg.Add(1)
|
|
|
|
go func(ctx context.Context, sem *semaphore.Weighted, wg *sync.WaitGroup, duty *attester.Duty, i int) {
|
|
|
|
defer wg.Done()
|
|
|
|
if err := sem.Acquire(ctx, 1); err != nil {
|
|
|
|
log.Error().Err(err).Msg("Failed to obtain semaphore")
|
2020-09-30 02:44:29 -07:00
|
|
|
return
|
2020-09-27 23:46:00 -07:00
|
|
|
}
|
|
|
|
defer sem.Release(1)
|
|
|
|
subscriptionInfoMutex.RLock()
|
|
|
|
info, exists := subscriptionInfo[duty.Slot()][duty.CommitteeIndices()[i]]
|
|
|
|
subscriptionInfoMutex.RUnlock()
|
2020-10-28 08:09:51 -07:00
|
|
|
if exists && info.IsAggregator {
|
2020-09-27 23:46:00 -07:00
|
|
|
// Already an aggregator for this slot/committee; don't need to go further.
|
|
|
|
return
|
|
|
|
}
|
|
|
|
isAggregator, signature, err := s.attestationAggregator.(attestationaggregator.IsAggregatorProvider).
|
|
|
|
IsAggregator(ctx,
|
|
|
|
duty.ValidatorIndices()[i],
|
|
|
|
duty.CommitteeIndices()[i],
|
|
|
|
duty.Slot(),
|
|
|
|
duty.CommitteeSize(duty.CommitteeIndices()[i]))
|
|
|
|
if err != nil {
|
2020-11-24 16:02:13 -08:00
|
|
|
log.Error().
|
|
|
|
Uint64("slot", uint64(duty.Slot())).
|
|
|
|
Uint64("validator_index", uint64(duty.ValidatorIndices()[i])).
|
|
|
|
Err(err).
|
|
|
|
Msg("Failed to calculate if validator is an aggregator")
|
2020-09-27 23:46:00 -07:00
|
|
|
return
|
|
|
|
}
|
2020-11-24 16:02:13 -08:00
|
|
|
// Obtain composite public key if available, otherwise standard public key.
|
|
|
|
account := accounts[duty.ValidatorIndices()[i]]
|
2021-07-17 23:34:43 -07:00
|
|
|
var pubKey phase0.BLSPubKey
|
2020-11-24 16:02:13 -08:00
|
|
|
if provider, isProvider := account.(e2wtypes.AccountCompositePublicKeyProvider); isProvider {
|
|
|
|
copy(pubKey[:], provider.CompositePublicKey().Marshal())
|
|
|
|
} else {
|
|
|
|
copy(pubKey[:], account.PublicKey().Marshal())
|
2020-09-27 23:46:00 -07:00
|
|
|
}
|
|
|
|
subscriptionInfoMutex.Lock()
|
|
|
|
if _, exists := subscriptionInfo[duty.Slot()]; !exists {
|
2021-07-17 23:34:43 -07:00
|
|
|
subscriptionInfo[duty.Slot()] = make(map[phase0.CommitteeIndex]*beaconcommitteesubscriber.Subscription)
|
2020-09-27 23:46:00 -07:00
|
|
|
}
|
|
|
|
subscriptionInfo[duty.Slot()][duty.CommitteeIndices()[i]] = &beaconcommitteesubscriber.Subscription{
|
2020-10-28 08:09:51 -07:00
|
|
|
Duty: &api.AttesterDuty{
|
|
|
|
PubKey: pubKey,
|
|
|
|
Slot: duty.Slot(),
|
|
|
|
ValidatorIndex: duty.ValidatorIndices()[i],
|
|
|
|
CommitteeIndex: duty.CommitteeIndices()[i],
|
|
|
|
CommitteeLength: duty.CommitteeSize(duty.CommitteeIndices()[i]),
|
|
|
|
CommitteesAtSlot: duty.CommitteesAtSlot(),
|
|
|
|
ValidatorCommitteeIndex: duty.ValidatorCommitteeIndices()[i],
|
|
|
|
},
|
|
|
|
IsAggregator: isAggregator,
|
|
|
|
Signature: signature,
|
2020-09-27 23:46:00 -07:00
|
|
|
}
|
|
|
|
subscriptionInfoMutex.Unlock()
|
|
|
|
}(ctx, sem, wg, duty, i)
|
|
|
|
}
|
|
|
|
}(ctx, sem, &wg, duty)
|
|
|
|
}
|
|
|
|
wg.Wait()
|
|
|
|
|
|
|
|
return subscriptionInfo, nil
|
|
|
|
}
|