-
-
Notifications
You must be signed in to change notification settings - Fork 1
/
aggregator.go
82 lines (71 loc) · 2.89 KB
/
aggregator.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
package validator
import (
"context"
ethpb "github.com/prysmaticlabs/ethereumapis/eth/v1alpha1"
"github.com/prysmaticlabs/prysm/beacon-chain/core/helpers"
"github.com/sirupsen/logrus"
"go.opencensus.io/trace"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
)
// SubmitAggregateAndProof is called by a validator when its assigned to be an aggregator.
// The beacon node will broadcast aggregated attestation and proof on the aggregator's behavior.
func (as *Server) SubmitAggregateAndProof(ctx context.Context, req *ethpb.AggregationRequest) (*ethpb.AggregationResponse, error) {
ctx, span := trace.StartSpan(ctx, "AggregatorServer.SubmitAggregation")
defer span.End()
span.AddAttributes(trace.Int64Attribute("slot", int64(req.Slot)))
if as.SyncChecker.Syncing() {
return nil, status.Errorf(codes.Unavailable, "Syncing to latest head, not ready to respond")
}
validatorIndex, exists, err := as.BeaconDB.ValidatorIndex(ctx, req.PublicKey)
if err != nil {
return nil, status.Errorf(codes.Internal, "Could not get validator index from DB: %v", err)
}
if !exists {
return nil, status.Error(codes.Internal, "Could not locate validator index in DB")
}
epoch := helpers.SlotToEpoch(req.Slot)
activeValidatorIndices, err := as.HeadFetcher.HeadValidatorsIndices(epoch)
if err != nil {
return nil, status.Errorf(codes.Internal, "Could not get validators: %v", err)
}
seed, err := as.HeadFetcher.HeadSeed(epoch)
if err != nil {
return nil, status.Errorf(codes.Internal, "Could not get seed: %v", err)
}
committee, err := helpers.BeaconCommittee(activeValidatorIndices, seed, req.Slot, req.CommitteeIndex)
if err != nil {
return nil, err
}
// Check if the validator is an aggregator
isAggregator, err := helpers.IsAggregator(uint64(len(committee)), req.SlotSignature)
if err != nil {
return nil, status.Errorf(codes.Internal, "Could not get aggregator status: %v", err)
}
if !isAggregator {
return nil, status.Errorf(codes.InvalidArgument, "Validator is not an aggregator")
}
// Retrieve the unaggregated attestation from pool.
aggregatedAtts := as.AttPool.AggregatedAttestationsBySlotIndex(req.Slot, req.CommitteeIndex)
for _, aggregatedAtt := range aggregatedAtts {
if ctx.Err() != nil {
return nil, ctx.Err()
}
if helpers.IsAggregated(aggregatedAtt) {
if err := as.P2P.Broadcast(ctx, ðpb.AggregateAttestationAndProof{
AggregatorIndex: validatorIndex,
SelectionProof: req.SlotSignature,
Aggregate: aggregatedAtt,
}); err != nil {
return nil, status.Errorf(codes.Internal, "Could not broadcast aggregated attestation: %v", err)
}
log.WithFields(logrus.Fields{
"slot": req.Slot,
"committeeIndex": req.CommitteeIndex,
"validatorIndex": validatorIndex,
"aggregatedCount": aggregatedAtt.AggregationBits.Count(),
}).Debug("Broadcasting aggregated attestation and proof")
}
}
return ðpb.AggregationResponse{}, nil
}