mirror of
https://gitlab.com/pulsechaincom/prysm-pulse.git
synced 2025-01-12 04:30:04 +00:00
14b3181e67
* more spanner additions * implement iface * begin implement * wrapped up spanner functions * rem interface * added in necessary comments * comments on enums * begin adding tests * plug in surround vote detection * saved indexed db implementation * finally plugin slashing for historical data * Small fixes * add in all gazelle * save incoming new functions * resolve todo * fix broken test channel item * tests passing when fixing certain arguments and setups * Add comment and change unimplemented * find surround * added in gazelle * gazz * feedback from shay * fixed up naming * Update * Add tests for detectSurroundVotes * Remove logs * Fix slasher test * formatting * Remove unneeded condition * Test indices better * fixing broken build * pass tests * skip tests * imports * Update slasher/detection/attestations/attestations_test.go * Update slasher/beaconclient/historical_data_retrieval_test.go * Address comments * Rename function * Add comment for future optimization * Fix comment Co-authored-by: Ivan Martinez <ivanthegreatdev@gmail.com>
77 lines
2.5 KiB
Go
77 lines
2.5 KiB
Go
package beaconclient
|
|
|
|
import (
|
|
"context"
|
|
"io"
|
|
|
|
ptypes "github.com/gogo/protobuf/types"
|
|
"go.opencensus.io/trace"
|
|
)
|
|
|
|
// 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 (bs *Service) receiveBlocks(ctx context.Context) {
|
|
ctx, span := trace.StartSpan(ctx, "beaconclient.receiveBlocks")
|
|
defer span.End()
|
|
stream, err := bs.beaconClient.StreamBlocks(ctx, &ptypes.Empty{})
|
|
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 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 {
|
|
log.WithError(err).Error("Could not receive block from beacon node")
|
|
}
|
|
log.WithField("slot", res.Block.Slot).Debug("Received block from beacon node")
|
|
// We send the received block over the block feed.
|
|
bs.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 (bs *Service) receiveAttestations(ctx context.Context) {
|
|
ctx, span := trace.StartSpan(ctx, "beaconclient.receiveAttestations")
|
|
defer span.End()
|
|
stream, err := bs.beaconClient.StreamIndexedAttestations(ctx, &ptypes.Empty{})
|
|
if err != nil {
|
|
log.WithError(err).Error("Failed to retrieve attestations stream")
|
|
return
|
|
}
|
|
for {
|
|
res, err := stream.Recv()
|
|
// If the stream is closed, we stop the loop.
|
|
if 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 attestations receiver")
|
|
return
|
|
}
|
|
if err != nil {
|
|
log.WithError(err).Error("Could not receive attestations from beacon node")
|
|
return
|
|
}
|
|
log.WithField("slot", res.Data.Slot).Debug("Received attestation from beacon node")
|
|
if err := bs.slasherDB.SaveIncomingIndexedAttestationByEpoch(ctx, res); err != nil {
|
|
log.WithError(err).Error("Could not save indexed attestation")
|
|
continue
|
|
}
|
|
// We send the received attestation over the attestation feed.
|
|
bs.attestationFeed.Send(res)
|
|
}
|
|
}
|