mirror of
https://gitlab.com/pulsechaincom/prysm-pulse.git
synced 2024-12-26 05:17:22 +00:00
7acd73e1fe
* extract SendBeaconBlocksByRangeRequest * adds extra processing * extract SendBeaconBlocksByRootRequest * extract into rpc_send_request.go * add tests * more tests * gazelle * remove redundant import * for consistency * for consistency Co-authored-by: prylabs-bulldozer[bot] <58059840+prylabs-bulldozer[bot]@users.noreply.github.com>
98 lines
3.2 KiB
Go
98 lines
3.2 KiB
Go
package sync
|
|
|
|
import (
|
|
"context"
|
|
|
|
libp2pcore "github.com/libp2p/go-libp2p-core"
|
|
"github.com/libp2p/go-libp2p-core/peer"
|
|
"github.com/pkg/errors"
|
|
ethpb "github.com/prysmaticlabs/ethereumapis/eth/v1alpha1"
|
|
"github.com/prysmaticlabs/prysm/beacon-chain/p2p/types"
|
|
"github.com/prysmaticlabs/prysm/shared/params"
|
|
)
|
|
|
|
// sendRecentBeaconBlocksRequest sends a recent beacon blocks request to a peer to get
|
|
// those corresponding blocks from that peer.
|
|
func (s *Service) sendRecentBeaconBlocksRequest(ctx context.Context, blockRoots *types.BeaconBlockByRootsReq, id peer.ID) error {
|
|
ctx, cancel := context.WithTimeout(ctx, respTimeout)
|
|
defer cancel()
|
|
|
|
_, err := SendBeaconBlocksByRootRequest(ctx, s.p2p, id, blockRoots, func(blk *ethpb.SignedBeaconBlock) error {
|
|
blkRoot, err := blk.Block.HashTreeRoot()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
s.pendingQueueLock.Lock()
|
|
s.insertBlockToPendingQueue(blk.Block.Slot, blk, blkRoot)
|
|
s.pendingQueueLock.Unlock()
|
|
return nil
|
|
})
|
|
return err
|
|
}
|
|
|
|
// beaconBlocksRootRPCHandler looks up the request blocks from the database from the given block roots.
|
|
func (s *Service) beaconBlocksRootRPCHandler(ctx context.Context, msg interface{}, stream libp2pcore.Stream) error {
|
|
defer func() {
|
|
if err := stream.Close(); err != nil {
|
|
log.WithError(err).Debug("Failed to close stream")
|
|
}
|
|
}()
|
|
ctx, cancel := context.WithTimeout(ctx, ttfbTimeout)
|
|
defer cancel()
|
|
SetRPCStreamDeadlines(stream)
|
|
log := log.WithField("handler", "beacon_blocks_by_root")
|
|
|
|
rawMsg, ok := msg.(*types.BeaconBlockByRootsReq)
|
|
if !ok {
|
|
return errors.New("message is not type BeaconBlockByRootsReq")
|
|
}
|
|
blockRoots := *rawMsg
|
|
if err := s.rateLimiter.validateRequest(stream, uint64(len(blockRoots))); err != nil {
|
|
return err
|
|
}
|
|
if len(blockRoots) == 0 {
|
|
// Add to rate limiter in the event no
|
|
// roots are requested.
|
|
s.rateLimiter.add(stream, 1)
|
|
resp, err := s.generateErrorResponse(responseCodeInvalidRequest, "no block roots provided in request")
|
|
if err != nil {
|
|
log.WithError(err).Debug("Failed to generate a response error")
|
|
} else if _, err := stream.Write(resp); err != nil {
|
|
log.WithError(err).Debugf("Failed to write to stream")
|
|
}
|
|
return errors.New("no block roots provided")
|
|
}
|
|
|
|
if uint64(len(blockRoots)) > params.BeaconNetworkConfig().MaxRequestBlocks {
|
|
resp, err := s.generateErrorResponse(responseCodeInvalidRequest, "requested more than the max block limit")
|
|
if err != nil {
|
|
log.WithError(err).Debug("Failed to generate a response error")
|
|
} else if _, err := stream.Write(resp); err != nil {
|
|
log.WithError(err).Debugf("Failed to write to stream")
|
|
}
|
|
return errors.New("requested more than the max block limit")
|
|
}
|
|
s.rateLimiter.add(stream, int64(len(blockRoots)))
|
|
|
|
for _, root := range blockRoots {
|
|
blk, err := s.db.Block(ctx, root)
|
|
if err != nil {
|
|
log.WithError(err).Debug("Failed to fetch block")
|
|
resp, err := s.generateErrorResponse(responseCodeServerError, genericError)
|
|
if err != nil {
|
|
log.WithError(err).Debug("Failed to generate a response error")
|
|
} else if _, err := stream.Write(resp); err != nil {
|
|
log.WithError(err).Debugf("Failed to write to stream")
|
|
}
|
|
return err
|
|
}
|
|
if blk == nil {
|
|
continue
|
|
}
|
|
if err := s.chunkWriter(stream, blk); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|