mirror of
https://gitlab.com/pulsechaincom/prysm-pulse.git
synced 2024-12-25 12:57:18 +00:00
4d1b5f42af
* Better comments for aggregator functions * Naming returned value
209 lines
6.6 KiB
Go
209 lines
6.6 KiB
Go
package client
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"time"
|
|
|
|
types "github.com/prysmaticlabs/eth2-types"
|
|
"github.com/prysmaticlabs/prysm/beacon-chain/core/helpers"
|
|
ethpb "github.com/prysmaticlabs/prysm/proto/eth/v1alpha1"
|
|
validatorpb "github.com/prysmaticlabs/prysm/proto/validator/accounts/v2"
|
|
"github.com/prysmaticlabs/prysm/shared/bls"
|
|
"github.com/prysmaticlabs/prysm/shared/params"
|
|
"github.com/prysmaticlabs/prysm/shared/slotutil"
|
|
"github.com/prysmaticlabs/prysm/shared/timeutils"
|
|
"github.com/prysmaticlabs/prysm/shared/traceutil"
|
|
"go.opencensus.io/trace"
|
|
"google.golang.org/grpc/codes"
|
|
"google.golang.org/grpc/status"
|
|
)
|
|
|
|
// SubmitAggregateAndProof submits the validator's signed slot signature to the beacon node
|
|
// via gRPC. Beacon node will verify the slot signature and determine if the validator is also
|
|
// an aggregator. If yes, then beacon node will broadcast aggregated signature and
|
|
// proof on the validator's behalf.
|
|
func (v *validator) SubmitAggregateAndProof(ctx context.Context, slot types.Slot, pubKey [48]byte) {
|
|
ctx, span := trace.StartSpan(ctx, "validator.SubmitAggregateAndProof")
|
|
defer span.End()
|
|
|
|
span.AddAttributes(trace.StringAttribute("validator", fmt.Sprintf("%#x", pubKey)))
|
|
fmtKey := fmt.Sprintf("%#x", pubKey[:])
|
|
|
|
duty, err := v.duty(pubKey)
|
|
if err != nil {
|
|
log.Errorf("Could not fetch validator assignment: %v", err)
|
|
if v.emitAccountMetrics {
|
|
ValidatorAggFailVec.WithLabelValues(fmtKey).Inc()
|
|
}
|
|
return
|
|
}
|
|
|
|
// Avoid sending beacon node duplicated aggregation requests.
|
|
k := validatorSubscribeKey(slot, duty.CommitteeIndex)
|
|
v.aggregatedSlotCommitteeIDCacheLock.Lock()
|
|
if v.aggregatedSlotCommitteeIDCache.Contains(k) {
|
|
v.aggregatedSlotCommitteeIDCacheLock.Unlock()
|
|
return
|
|
}
|
|
v.aggregatedSlotCommitteeIDCache.Add(k, true)
|
|
v.aggregatedSlotCommitteeIDCacheLock.Unlock()
|
|
|
|
slotSig, err := v.signSlotWithSelectionProof(ctx, pubKey, slot)
|
|
if err != nil {
|
|
log.Errorf("Could not sign slot: %v", err)
|
|
if v.emitAccountMetrics {
|
|
ValidatorAggFailVec.WithLabelValues(fmtKey).Inc()
|
|
}
|
|
return
|
|
}
|
|
|
|
// As specified in spec, an aggregator should wait until two thirds of the way through slot
|
|
// to broadcast the best aggregate to the global aggregate channel.
|
|
// https://github.com/ethereum/eth2.0-specs/blob/v0.9.3/specs/validator/0_beacon-chain-validator.md#broadcast-aggregate
|
|
v.waitToSlotTwoThirds(ctx, slot)
|
|
|
|
res, err := v.validatorClient.SubmitAggregateSelectionProof(ctx, ðpb.AggregateSelectionRequest{
|
|
Slot: slot,
|
|
CommitteeIndex: duty.CommitteeIndex,
|
|
PublicKey: pubKey[:],
|
|
SlotSignature: slotSig,
|
|
})
|
|
if err != nil {
|
|
s, ok := status.FromError(err)
|
|
if ok && s.Code() == codes.NotFound {
|
|
log.WithField("slot", slot).WithError(err).Warn("No attestations to aggregate")
|
|
} else {
|
|
log.WithField("slot", slot).WithError(err).Error("Could not submit slot signature to beacon node")
|
|
if v.emitAccountMetrics {
|
|
ValidatorAggFailVec.WithLabelValues(fmtKey).Inc()
|
|
}
|
|
}
|
|
|
|
return
|
|
}
|
|
|
|
sig, err := v.aggregateAndProofSig(ctx, pubKey, res.AggregateAndProof)
|
|
if err != nil {
|
|
log.Errorf("Could not sign aggregate and proof: %v", err)
|
|
return
|
|
}
|
|
_, err = v.validatorClient.SubmitSignedAggregateSelectionProof(ctx, ðpb.SignedAggregateSubmitRequest{
|
|
SignedAggregateAndProof: ðpb.SignedAggregateAttestationAndProof{
|
|
Message: res.AggregateAndProof,
|
|
Signature: sig,
|
|
},
|
|
})
|
|
if err != nil {
|
|
log.Errorf("Could not submit signed aggregate and proof to beacon node: %v", err)
|
|
if v.emitAccountMetrics {
|
|
ValidatorAggFailVec.WithLabelValues(fmtKey).Inc()
|
|
}
|
|
return
|
|
}
|
|
|
|
if err := v.addIndicesToLog(duty); err != nil {
|
|
log.Errorf("Could not add aggregator indices to logs: %v", err)
|
|
if v.emitAccountMetrics {
|
|
ValidatorAggFailVec.WithLabelValues(fmtKey).Inc()
|
|
}
|
|
return
|
|
}
|
|
if v.emitAccountMetrics {
|
|
ValidatorAggSuccessVec.WithLabelValues(fmtKey).Inc()
|
|
}
|
|
|
|
}
|
|
|
|
// Signs input slot with domain selection proof. This is used to create the signature for aggregator selection.
|
|
func (v *validator) signSlotWithSelectionProof(ctx context.Context, pubKey [48]byte, slot types.Slot) (signature []byte, error error) {
|
|
domain, err := v.domainData(ctx, helpers.SlotToEpoch(slot), params.BeaconConfig().DomainSelectionProof[:])
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
var sig bls.Signature
|
|
sszUint := types.SSZUint64(slot)
|
|
root, err := helpers.ComputeSigningRoot(&sszUint, domain.SignatureDomain)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
sig, err = v.keyManager.Sign(ctx, &validatorpb.SignRequest{
|
|
PublicKey: pubKey[:],
|
|
SigningRoot: root[:],
|
|
SignatureDomain: domain.SignatureDomain,
|
|
Object: &validatorpb.SignRequest_Slot{Slot: slot},
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return sig.Marshal(), nil
|
|
}
|
|
|
|
// waitToSlotTwoThirds waits until two third through the current slot period
|
|
// such that any attestations from this slot have time to reach the beacon node
|
|
// before creating the aggregated attestation.
|
|
func (v *validator) waitToSlotTwoThirds(ctx context.Context, slot types.Slot) {
|
|
ctx, span := trace.StartSpan(ctx, "validator.waitToSlotTwoThirds")
|
|
defer span.End()
|
|
|
|
oneThird := slotutil.DivideSlotBy(3 /* one third of slot duration */)
|
|
twoThird := oneThird + oneThird
|
|
delay := twoThird
|
|
|
|
startTime := slotutil.SlotStartTime(v.genesisTime, slot)
|
|
finalTime := startTime.Add(delay)
|
|
wait := timeutils.Until(finalTime)
|
|
if wait <= 0 {
|
|
return
|
|
}
|
|
t := time.NewTimer(wait)
|
|
defer t.Stop()
|
|
select {
|
|
case <-ctx.Done():
|
|
traceutil.AnnotateError(span, ctx.Err())
|
|
return
|
|
case <-t.C:
|
|
return
|
|
}
|
|
}
|
|
|
|
// This returns the signature of validator signing over aggregate and
|
|
// proof object.
|
|
func (v *validator) aggregateAndProofSig(ctx context.Context, pubKey [48]byte, agg *ethpb.AggregateAttestationAndProof) ([]byte, error) {
|
|
d, err := v.domainData(ctx, helpers.SlotToEpoch(agg.Aggregate.Data.Slot), params.BeaconConfig().DomainAggregateAndProof[:])
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var sig bls.Signature
|
|
root, err := helpers.ComputeSigningRoot(agg, d.SignatureDomain)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
sig, err = v.keyManager.Sign(ctx, &validatorpb.SignRequest{
|
|
PublicKey: pubKey[:],
|
|
SigningRoot: root[:],
|
|
SignatureDomain: d.SignatureDomain,
|
|
Object: &validatorpb.SignRequest_AggregateAttestationAndProof{AggregateAttestationAndProof: agg},
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return sig.Marshal(), nil
|
|
}
|
|
|
|
func (v *validator) addIndicesToLog(duty *ethpb.DutiesResponse_Duty) error {
|
|
v.attLogsLock.Lock()
|
|
defer v.attLogsLock.Unlock()
|
|
|
|
for _, log := range v.attLogs {
|
|
if duty.CommitteeIndex == log.data.CommitteeIndex {
|
|
log.aggregatorIndices = append(log.aggregatorIndices, duty.ValidatorIndex)
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|