prysm-pulse/beacon-chain/sync/rpc_hello.go
Preston Van Loon 78bf39aff7
sync RPC: Hello handler (#3216)
* checkpoint

* checkpoint

* varint prefix for ssz

* move the encoding API around a little bit to support reader writer

* add a simple test for the happy path subscribe

* move wait timeout to testutil

* Add inverted topic mapping

* Add varint prefixing to ssz network encoder

* fix spacing

* fix comments

* fix comments

* make anon methods more clear

* clean up log fields

* move topic mapping, reformat TODOs, get ready for brutal team review

* lint

* lint

* lint

* Update beacon-chain/p2p/gossip_topic_mappings.go

Co-Authored-By: Nishant Das <nishdas93@gmail.com>

* PR feedback

* PR feedback

* PR feedback

* PR feedback

* PR feedback

* basic test with a hardcoded fork choice

* updated beacon config to use genesis fork version

* checkpoint on hello handler

* create a testing db method that can be used with the new database interface

* lint

* lint

* PR feedback

* checkpoint

* passing tests

* comments, errors

* comments, errors

* remove swarm

* Update WORKSPACE

* Update WORKSPACE

* merge

* lint

* lint

* touch

* touch

* imports
2019-08-16 16:03:11 -04:00

71 lines
2.3 KiB
Go

package sync
import (
"bytes"
"context"
"time"
"github.com/gogo/protobuf/proto"
libp2pcore "github.com/libp2p/go-libp2p-core"
pb "github.com/prysmaticlabs/prysm/proto/beacon/p2p/v1"
"github.com/prysmaticlabs/prysm/shared/params"
)
// helloRPCHandler reads the incoming Hello RPC from the peer and responds with our version of a hello message.
// This handler will disconnect any peer that does not match our fork version.
func (r *RegularSync) helloRPCHandler(ctx context.Context, msg proto.Message, stream libp2pcore.Stream) error {
defer stream.Close()
ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
defer cancel()
setRPCStreamDeadlines(stream)
log := log.WithField("rpc", "hello")
m := msg.(*pb.Hello)
if !bytes.Equal(params.BeaconConfig().GenesisForkVersion, m.ForkVersion) {
resp, err := r.generateErrorResponse(responseCodeInvalidRequest, errWrongForkVersion.Error())
if err != nil {
log.WithError(err).Error("Failed to generate a response error")
} else {
if _, err := stream.Write(resp); err != nil {
log.WithError(err).Errorf("Failed to write to stream")
}
}
stream.Close() // Close before disconnecting.
// Add a short delay to allow the stream to flush before closing the connection.
// There is still a chance that the peer won't receive the message.
time.Sleep(50 * time.Millisecond)
if err := r.p2p.Disconnect(stream.Conn().RemotePeer()); err != nil {
log.WithError(err).Error("Failed to disconnect from peer")
}
return errWrongForkVersion
}
r.p2p.AddHandshake(stream.Conn().RemotePeer(), m)
state, err := r.db.HeadState(ctx)
if err != nil {
log.WithError(err).Error("Failed to get head state")
resp, err := r.generateErrorResponse(responseCodeServerError, genericError)
if _, err := stream.Write(resp); err != nil {
log.WithError(err).Error("Failed to write to stream")
}
return err
}
resp := &pb.Hello{
ForkVersion: params.BeaconConfig().GenesisForkVersion,
FinalizedRoot: state.FinalizedCheckpoint.Root,
FinalizedEpoch: state.FinalizedCheckpoint.Epoch,
HeadRoot: state.BlockRoots[state.Slot%params.BeaconConfig().SlotsPerHistoricalRoot],
HeadSlot: state.Slot,
}
if _, err := stream.Write([]byte{responseCodeSuccess}); err != nil {
log.WithError(err).Error("Failed to write to stream")
}
_, err = r.p2p.Encoding().Encode(stream, resp)
return err
}