Address Review of v0.12

beacon-chain/p2p/broadcaster.go

@@ -18,12 +18,16 @@ import (
 // GossipTypeMapping.
 var ErrMessageNotMapped = errors.New("message type is not mapped to a PubSub topic")
 
+const attestationSubnetTopicFormat = "/eth2/%x/beacon_attestation_%d"
+
 // Broadcast a message to the p2p network.
 func (s *Service) Broadcast(ctx context.Context, msg proto.Message) error {
 	ctx, span := trace.StartSpan(ctx, "p2p.Broadcast")
 	defer span.End()
 	forkDigest, err := s.forkDigest()
 	if err != nil {
+		err := errors.Wrap(err, "could not retrieve fork digest")
+		traceutil.AnnotateError(span, err)
 		return err
 	}
 
@@ -32,29 +36,7 @@ func (s *Service) Broadcast(ctx context.Context, msg proto.Message) error {
 		traceutil.AnnotateError(span, ErrMessageNotMapped)
 		return ErrMessageNotMapped
 	}
-	topic = fmt.Sprintf(topic, forkDigest)
-
-	span.AddAttributes(trace.StringAttribute("topic", topic))
-
-	buf := new(bytes.Buffer)
-	if _, err := s.Encoding().EncodeGossip(buf, msg); err != nil {
-		err := errors.Wrap(err, "could not encode message")
-		traceutil.AnnotateError(span, err)
-		return err
-	}
-
-	if span.IsRecordingEvents() {
-		id := hashutil.FastSum64(buf.Bytes())
-		messageLen := int64(buf.Len())
-		span.AddMessageSendEvent(int64(id), messageLen /*uncompressed*/, messageLen /*compressed*/)
-	}
-
-	if err := s.pubsub.Publish(topic+s.Encoding().ProtocolSuffix(), buf.Bytes()); err != nil {
-		err := errors.Wrap(err, "could not publish message")
-		traceutil.AnnotateError(span, err)
-		return err
-	}
-	return nil
+	return s.broadcastObject(ctx, msg, fmt.Sprintf(topic, forkDigest))
 }
 
 // BroadcastAttestation broadcasts an attestation to the p2p network.
@@ -63,13 +45,22 @@ func (s *Service) BroadcastAttestation(ctx context.Context, subnet uint64, att *
 	defer span.End()
 	forkDigest, err := s.forkDigest()
 	if err != nil {
+		err := errors.Wrap(err, "could not retrieve fork digest")
+		traceutil.AnnotateError(span, err)
 		return err
 	}
-	topic := attestationToTopic(subnet, forkDigest)
+	return s.broadcastObject(ctx, att, attestationToTopic(subnet, forkDigest))
+}
+
+// method to broadcast messages to other peers in our gossip mesh.
+func (s *Service) broadcastObject(ctx context.Context, obj interface{}, topic string) error {
+	_, span := trace.StartSpan(ctx, "p2p.broadcastObject")
+	defer span.End()
+
 	span.AddAttributes(trace.StringAttribute("topic", topic))
 
 	buf := new(bytes.Buffer)
-	if _, err := s.Encoding().EncodeGossip(buf, att); err != nil {
+	if _, err := s.Encoding().EncodeGossip(buf, obj); err != nil {
 		err := errors.Wrap(err, "could not encode message")
 		traceutil.AnnotateError(span, err)
 		return err
@@ -89,8 +80,6 @@ func (s *Service) BroadcastAttestation(ctx context.Context, subnet uint64, att *
 	return nil
 }
 
-const attestationSubnetTopicFormat = "/eth2/%x/beacon_attestation_%d"
-
 func attestationToTopic(subnet uint64, forkDigest [4]byte) string {
 	return fmt.Sprintf(attestationSubnetTopicFormat, forkDigest, subnet)
 }

beacon-chain/rpc/validator/attester.go

@@ -148,6 +148,9 @@ func (vs *Server) GetAttestationData(ctx context.Context, req *ethpb.Attestation
 // ProposeAttestation is a function called by an attester to vote
 // on a block via an attestation object as defined in the Ethereum Serenity specification.
 func (vs *Server) ProposeAttestation(ctx context.Context, att *ethpb.Attestation) (*ethpb.AttestResponse, error) {
+	ctx, span := trace.StartSpan(ctx, "AttesterServer.ProposeAttestation")
+	defer span.End()
+
 	if _, err := bls.SignatureFromBytes(att.Signature); err != nil {
 		return nil, status.Error(codes.InvalidArgument, "Incorrect attestation signature")
 	}
@@ -195,11 +198,14 @@ func (vs *Server) ProposeAttestation(ctx context.Context, att *ethpb.Attestation
 
 // SubscribeCommitteeSubnets subscribes to the committee ID subnet given subscribe request.
 func (vs *Server) SubscribeCommitteeSubnets(ctx context.Context, req *ethpb.CommitteeSubnetsSubscribeRequest) (*ptypes.Empty, error) {
+	ctx, span := trace.StartSpan(ctx, "AttesterServer.SubscribeCommitteeSubnets")
+	defer span.End()
+
 	if len(req.Slots) != len(req.CommitteeIds) && len(req.CommitteeIds) != len(req.IsAggregator) {
 		return nil, status.Error(codes.InvalidArgument, "request fields are not the same length")
 	}
 	if len(req.Slots) == 0 {
-		return &ptypes.Empty{}, nil
+		return nil, status.Error(codes.InvalidArgument, "no attester slots provided")
 	}
 
 	fetchValsLen := func(slot uint64) (uint64, error) {
@@ -211,19 +217,20 @@ func (vs *Server) SubscribeCommitteeSubnets(ctx context.Context, req *ethpb.Comm
 		return uint64(len(vals)), nil
 	}
 
-	// run first request
+	// Request the head validator indices of epoch represented by the first requested
+	// slot.
 	currValsLen, err := fetchValsLen(req.Slots[0])
 	if err != nil {
-		return nil, err
+		return nil, status.Errorf(codes.Internal, "Could not retrieve head validator length: %v", err)
 	}
 	currEpoch := helpers.SlotToEpoch(req.Slots[0])
 
 	for i := 0; i < len(req.Slots); i++ {
-		// if epoch has changed, re-request active validators
+		// If epoch has changed, re-request active validators length
 		if currEpoch != helpers.SlotToEpoch(req.Slots[i]) {
 			currValsLen, err = fetchValsLen(req.Slots[i])
 			if err != nil {
-				return nil, err
+				return nil, status.Errorf(codes.Internal, "Could not retrieve head validator length: %v", err)
 			}
 			currEpoch = helpers.SlotToEpoch(req.Slots[i])
 		}