prysm-pulse/slasher/beaconclient/receivers.go
2021-07-21 21:34:07 +00:00

217 lines
6.8 KiB
Go

package beaconclient
import (
"context"
"errors"
"fmt"
"io"
"time"
ethpb "github.com/prysmaticlabs/prysm/proto/prysm/v1alpha1"
"github.com/prysmaticlabs/prysm/shared/slotutil"
"github.com/sirupsen/logrus"
"go.opencensus.io/trace"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/connectivity"
"google.golang.org/grpc/status"
"google.golang.org/protobuf/types/known/emptypb"
)
// reconnectPeriod is the frequency that we try to restart our
// streams when the beacon chain is node does not respond.
var reconnectPeriod = 5 * time.Second
// ReceiveBlocks starts a gRPC client stream listener to obtain
// blocks from the beacon node. Upon receiving a block, the service
// broadcasts it to a feed for other services in slasher to subscribe to.
func (s *Service) ReceiveBlocks(ctx context.Context) {
ctx, span := trace.StartSpan(ctx, "beaconclient.ReceiveBlocks")
defer span.End()
stream, err := s.cfg.BeaconClient.StreamBlocks(ctx, &ethpb.StreamBlocksRequest{} /* Prefers unverified block to catch slashing */)
if err != nil {
log.WithError(err).Error("Failed to retrieve blocks stream")
return
}
for {
res, err := stream.Recv()
// If the stream is closed, we stop the loop.
if errors.Is(err, io.EOF) {
break
}
// If context is canceled we stop the loop.
if ctx.Err() == context.Canceled {
log.WithError(ctx.Err()).Error("Context canceled - shutting down blocks receiver")
return
}
if err != nil {
if e, ok := status.FromError(err); ok {
switch e.Code() {
case codes.Canceled, codes.Internal, codes.Unavailable:
log.WithError(err).Infof("Trying to restart connection. rpc status: %v", e.Code())
err = s.restartBeaconConnection(ctx)
if err != nil {
log.WithError(err).Error("Could not restart beacon connection")
return
}
stream, err = s.cfg.BeaconClient.StreamBlocks(ctx, &ethpb.StreamBlocksRequest{} /* Prefers unverified block to catch slashing */)
if err != nil {
log.WithError(err).Error("Could not restart block stream")
return
}
log.Info("Block stream restarted...")
default:
log.WithError(err).Errorf("Could not receive block from beacon node. rpc status: %v", e.Code())
return
}
} else {
log.WithError(err).Error("Could not receive blocks from beacon node")
return
}
}
if res == nil {
continue
}
root, err := res.Block.HashTreeRoot()
if err != nil {
log.WithError(err).Error("Could not hash block")
return
}
log.WithFields(logrus.Fields{
"slot": res.Block.Slot,
"proposer_index": res.Block.ProposerIndex,
"root": fmt.Sprintf("%#x...", root[:8]),
}).Info("Received block from beacon node")
// We send the received block over the block feed.
s.blockFeed.Send(res)
}
}
// ReceiveAttestations starts a gRPC client stream listener to obtain
// attestations from the beacon node. Upon receiving an attestation, the service
// broadcasts it to a feed for other services in slasher to subscribe to.
func (s *Service) ReceiveAttestations(ctx context.Context) {
ctx, span := trace.StartSpan(ctx, "beaconclient.ReceiveAttestations")
defer span.End()
stream, err := s.cfg.BeaconClient.StreamIndexedAttestations(ctx, &emptypb.Empty{})
if err != nil {
log.WithError(err).Error("Failed to retrieve attestations stream")
return
}
go s.collectReceivedAttestations(ctx)
for {
res, err := stream.Recv()
// If the stream is closed, we stop the loop.
if errors.Is(err, io.EOF) {
log.Info("Attestation stream closed")
break
}
// If context is canceled we stop the loop.
if ctx.Err() == context.Canceled {
log.WithError(ctx.Err()).Error("Context canceled - shutting down attestations receiver")
return
}
if err != nil {
if e, ok := status.FromError(err); ok {
switch e.Code() {
case codes.Canceled, codes.Internal, codes.Unavailable:
log.WithError(err).Infof("Trying to restart connection. rpc status: %v", e.Code())
err = s.restartBeaconConnection(ctx)
if err != nil {
log.WithError(err).Error("Could not restart beacon connection")
return
}
stream, err = s.cfg.BeaconClient.StreamIndexedAttestations(ctx, &emptypb.Empty{})
if err != nil {
log.WithError(err).Error("Could not restart attestation stream")
return
}
log.Info("Attestation stream restarted...")
default:
log.WithError(err).Errorf("Could not receive attestations from beacon node. rpc status: %v", e.Code())
return
}
} else {
log.WithError(err).Error("Could not receive attestations from beacon node")
return
}
}
if res == nil {
continue
}
s.receivedAttestationsBuffer <- res
}
}
func (s *Service) collectReceivedAttestations(ctx context.Context) {
ctx, span := trace.StartSpan(ctx, "beaconclient.collectReceivedAttestations")
defer span.End()
var atts []*ethpb.IndexedAttestation
halfSlot := slotutil.DivideSlotBy(2 /* 1/2 slot duration */)
ticker := time.NewTicker(halfSlot)
defer ticker.Stop()
for {
select {
case <-ticker.C:
if len(atts) > 0 {
s.collectedAttestationsBuffer <- atts
atts = []*ethpb.IndexedAttestation{}
}
case att := <-s.receivedAttestationsBuffer:
atts = append(atts, att)
case collectedAtts := <-s.collectedAttestationsBuffer:
if err := s.cfg.SlasherDB.SaveIndexedAttestations(ctx, collectedAtts); err != nil {
log.WithError(err).Error("Could not save indexed attestation")
continue
}
log.WithFields(logrus.Fields{
"amountSaved": len(collectedAtts),
"slot": collectedAtts[0].Data.Slot,
}).Info("Attestations saved to slasher DB")
slasherNumAttestationsReceived.Add(float64(len(collectedAtts)))
// After saving, we send the received attestation over the attestation feed.
for _, att := range collectedAtts {
log.WithFields(logrus.Fields{
"slot": att.Data.Slot,
"indices": att.AttestingIndices,
}).Debug("Sending attestation to detection service")
s.attestationFeed.Send(att)
}
case <-ctx.Done():
return
}
}
}
func (s *Service) restartBeaconConnection(ctx context.Context) error {
ticker := time.NewTicker(reconnectPeriod)
defer ticker.Stop()
for {
select {
case <-ticker.C:
if s.conn.GetState() == connectivity.TransientFailure || s.conn.GetState() == connectivity.Idle {
log.Debugf("Connection status %v", s.conn.GetState())
log.Info("Beacon node is still down")
continue
}
s, err := s.cfg.NodeClient.GetSyncStatus(ctx, &emptypb.Empty{})
if err != nil {
log.WithError(err).Error("Could not fetch sync status")
continue
}
if s == nil || s.Syncing {
log.Info("Waiting for beacon node to be fully synced...")
continue
}
log.Info("Beacon node is fully synced")
return nil
case <-ctx.Done():
log.Debug("Context closed, exiting reconnect routine")
return errors.New("context closed, no longer attempting to restart stream")
}
}
}