mirror of
https://gitlab.com/pulsechaincom/prysm-pulse.git
synced 2024-12-26 05:17:22 +00:00
2428880058
* Update go-libp2p to 0.12.0 go-libp2p 0.12.0 made some significant changes to the stream interfaces around stream closing: * Close now closes in both directions and frees the stream. However, unlike FullClose did, it doesn't _wait_ for the remote peer to respond with an EOF. * To close for writing, call CloseWrite (like one would on a TCP connection, etc.). This patch: * Replaces calls to FullClose with Close where appropriate. * Replaces calls to Close with CloseWrite where appropriate. * Removes redundant Close calls. * Calls Reset to where appropriate to indicate that the request/response was aborted. Unlike Close, this will not flush and will not cause the remote peer to read an EOF. Instead, the remote peer will read an ErrReset error. * Ensures we always either close or reset streams. Send wasn't closing the stream on some error paths. * Now that stream closing is async, we explicitly wait for a response when "hanging up" on a peer (so we don't hang up before they receive our response/goodbye message). * update bazel * Gazelle * revert unintentional bazel workspace change * appease an overzealous linter * update to latest * Refactor encoder * gazelle * Gazelle Co-authored-by: Preston Van Loon <preston@prysmaticlabs.com> Co-authored-by: Nishant Das <nishdas93@gmail.com> Co-authored-by: Raul Jordan <raul@prysmaticlabs.com>
81 lines
2.6 KiB
Go
81 lines
2.6 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()
|
|
if err := s.insertBlockToPendingQueue(blk.Block.Slot, blk, blkRoot); err != nil {
|
|
return err
|
|
}
|
|
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 {
|
|
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)
|
|
s.writeErrorResponseToStream(responseCodeInvalidRequest, "no block roots provided in request", stream)
|
|
return errors.New("no block roots provided")
|
|
}
|
|
|
|
if uint64(len(blockRoots)) > params.BeaconNetworkConfig().MaxRequestBlocks {
|
|
s.writeErrorResponseToStream(responseCodeInvalidRequest, "requested more than the max block limit", 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("Could not fetch block")
|
|
s.writeErrorResponseToStream(responseCodeServerError, types.ErrGeneric.Error(), stream)
|
|
return err
|
|
}
|
|
if blk == nil {
|
|
continue
|
|
}
|
|
if err := s.chunkWriter(stream, blk); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
closeStream(stream, log)
|
|
return nil
|
|
}
|