mirror of
https://gitlab.com/pulsechaincom/prysm-pulse.git
synced 2024-12-23 11:57:18 +00:00
16bbf5602f
* builds * move block to consensus-types Co-authored-by: prylabs-bulldozer[bot] <58059840+prylabs-bulldozer[bot]@users.noreply.github.com>
158 lines
5.1 KiB
Go
158 lines
5.1 KiB
Go
package sync
|
|
|
|
import (
|
|
"context"
|
|
|
|
libp2pcore "github.com/libp2p/go-libp2p-core"
|
|
"github.com/libp2p/go-libp2p-core/peer"
|
|
"github.com/pkg/errors"
|
|
"github.com/prysmaticlabs/go-bitfield"
|
|
"github.com/prysmaticlabs/prysm/beacon-chain/blockchain"
|
|
"github.com/prysmaticlabs/prysm/beacon-chain/core/signing"
|
|
"github.com/prysmaticlabs/prysm/beacon-chain/p2p"
|
|
"github.com/prysmaticlabs/prysm/beacon-chain/p2p/types"
|
|
"github.com/prysmaticlabs/prysm/config/params"
|
|
"github.com/prysmaticlabs/prysm/consensus-types/wrapper"
|
|
"github.com/prysmaticlabs/prysm/encoding/bytesutil"
|
|
"github.com/prysmaticlabs/prysm/network/forks"
|
|
pb "github.com/prysmaticlabs/prysm/proto/prysm/v1alpha1"
|
|
"github.com/prysmaticlabs/prysm/proto/prysm/v1alpha1/metadata"
|
|
"github.com/prysmaticlabs/prysm/runtime/version"
|
|
"github.com/prysmaticlabs/prysm/time/slots"
|
|
)
|
|
|
|
// metaDataHandler reads the incoming metadata rpc request from the peer.
|
|
func (s *Service) metaDataHandler(_ context.Context, _ interface{}, stream libp2pcore.Stream) error {
|
|
SetRPCStreamDeadlines(stream)
|
|
|
|
if err := s.rateLimiter.validateRequest(stream, 1); err != nil {
|
|
return err
|
|
}
|
|
s.rateLimiter.add(stream, 1)
|
|
|
|
if s.cfg.p2p.Metadata() == nil || s.cfg.p2p.Metadata().IsNil() {
|
|
nilErr := errors.New("nil metadata stored for host")
|
|
resp, err := s.generateErrorResponse(responseCodeServerError, types.ErrGeneric.Error())
|
|
if err != nil {
|
|
log.WithError(err).Debug("Could not generate a response error")
|
|
} else if _, err := stream.Write(resp); err != nil {
|
|
log.WithError(err).Debug("Could not write to stream")
|
|
}
|
|
return nilErr
|
|
}
|
|
_, _, streamVersion, err := p2p.TopicDeconstructor(string(stream.Protocol()))
|
|
if err != nil {
|
|
resp, genErr := s.generateErrorResponse(responseCodeServerError, types.ErrGeneric.Error())
|
|
if genErr != nil {
|
|
log.WithError(genErr).Debug("Could not generate a response error")
|
|
} else if _, wErr := stream.Write(resp); wErr != nil {
|
|
log.WithError(wErr).Debug("Could not write to stream")
|
|
}
|
|
return err
|
|
}
|
|
currMd := s.cfg.p2p.Metadata()
|
|
switch streamVersion {
|
|
case p2p.SchemaVersionV1:
|
|
// We have a v1 metadata object saved locally, so we
|
|
// convert it back to a v0 metadata object.
|
|
if currMd.Version() != version.Phase0 {
|
|
currMd = wrapper.WrappedMetadataV0(
|
|
&pb.MetaDataV0{
|
|
Attnets: currMd.AttnetsBitfield(),
|
|
SeqNumber: currMd.SequenceNumber(),
|
|
})
|
|
}
|
|
case p2p.SchemaVersionV2:
|
|
// We have a v0 metadata object saved locally, so we
|
|
// convert it to a v1 metadata object.
|
|
if currMd.Version() != version.Altair {
|
|
currMd = wrapper.WrappedMetadataV1(
|
|
&pb.MetaDataV1{
|
|
Attnets: currMd.AttnetsBitfield(),
|
|
SeqNumber: currMd.SequenceNumber(),
|
|
Syncnets: bitfield.Bitvector4{byte(0x00)},
|
|
})
|
|
}
|
|
}
|
|
if _, err := stream.Write([]byte{responseCodeSuccess}); err != nil {
|
|
return err
|
|
}
|
|
_, err = s.cfg.p2p.Encoding().EncodeWithMaxLength(stream, currMd)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
closeStream(stream, log)
|
|
return nil
|
|
}
|
|
|
|
func (s *Service) sendMetaDataRequest(ctx context.Context, id peer.ID) (metadata.Metadata, error) {
|
|
ctx, cancel := context.WithTimeout(ctx, respTimeout)
|
|
defer cancel()
|
|
|
|
topic, err := p2p.TopicFromMessage(p2p.MetadataMessageName, slots.ToEpoch(s.cfg.chain.CurrentSlot()))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
stream, err := s.cfg.p2p.Send(ctx, new(interface{}), topic, id)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer closeStream(stream, log)
|
|
code, errMsg, err := ReadStatusCode(stream, s.cfg.p2p.Encoding())
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if code != 0 {
|
|
s.cfg.p2p.Peers().Scorers().BadResponsesScorer().Increment(stream.Conn().RemotePeer())
|
|
return nil, errors.New(errMsg)
|
|
}
|
|
valRoot := s.cfg.chain.GenesisValidatorsRoot()
|
|
rpcCtx, err := forks.ForkDigestFromEpoch(slots.ToEpoch(s.cfg.chain.CurrentSlot()), valRoot[:])
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
msg, err := extractMetaDataType(rpcCtx[:], s.cfg.chain)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
// Defensive check to ensure valid objects are being sent.
|
|
topicVersion := ""
|
|
switch msg.Version() {
|
|
case version.Phase0:
|
|
topicVersion = p2p.SchemaVersionV1
|
|
case version.Altair:
|
|
topicVersion = p2p.SchemaVersionV2
|
|
}
|
|
if err := validateVersion(topicVersion, stream); err != nil {
|
|
return nil, err
|
|
}
|
|
if err := s.cfg.p2p.Encoding().DecodeWithMaxLength(stream, msg); err != nil {
|
|
return nil, err
|
|
}
|
|
return msg, nil
|
|
}
|
|
|
|
func extractMetaDataType(digest []byte, chain blockchain.ChainInfoFetcher) (metadata.Metadata, error) {
|
|
if len(digest) == 0 {
|
|
mdFunc, ok := types.MetaDataMap[bytesutil.ToBytes4(params.BeaconConfig().GenesisForkVersion)]
|
|
if !ok {
|
|
return nil, errors.New("no metadata type exists for the genesis fork version.")
|
|
}
|
|
return mdFunc(), nil
|
|
}
|
|
if len(digest) != forkDigestLength {
|
|
return nil, errors.Errorf("invalid digest returned, wanted a length of %d but received %d", forkDigestLength, len(digest))
|
|
}
|
|
vRoot := chain.GenesisValidatorsRoot()
|
|
for k, mdFunc := range types.MetaDataMap {
|
|
rDigest, err := signing.ComputeForkDigest(k[:], vRoot[:])
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if rDigest == bytesutil.ToBytes4(digest) {
|
|
return mdFunc(), nil
|
|
}
|
|
}
|
|
return nil, errors.New("no valid digest matched")
|
|
}
|