prysm-pulse/beacon-chain/sync/rpc_goodbye.go
Steven Allen 2428880058
Update go-libp2p to 0.12.0 (#8015)
* 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>
2020-12-14 17:22:25 +00:00

123 lines
4.1 KiB
Go

package sync
import (
"context"
"fmt"
"time"
libp2pcore "github.com/libp2p/go-libp2p-core"
"github.com/libp2p/go-libp2p-core/peer"
"github.com/prysmaticlabs/prysm/beacon-chain/p2p"
"github.com/prysmaticlabs/prysm/beacon-chain/p2p/types"
"github.com/sirupsen/logrus"
)
var backOffTime = map[types.SSZUint64]time.Duration{
// Do not dial peers which are from a different/unverifiable
// network.
types.GoodbyeCodeWrongNetwork: 24 * time.Hour,
types.GoodbyeCodeUnableToVerifyNetwork: 24 * time.Hour,
// If local peer is banned, we back off for
// 2 hours to let the remote peer score us
// back up again.
types.GoodbyeCodeBadScore: 2 * time.Hour,
types.GoodbyeCodeBanned: 2 * time.Hour,
types.GoodbyeCodeClientShutdown: 1 * time.Hour,
// Wait 5 minutes before dialing a peer who is
// 'full'
types.GoodbyeCodeTooManyPeers: 5 * time.Minute,
types.GoodbyeCodeGenericError: 2 * time.Minute,
}
// goodbyeRPCHandler reads the incoming goodbye rpc message from the peer.
func (s *Service) goodbyeRPCHandler(_ context.Context, msg interface{}, stream libp2pcore.Stream) error {
SetRPCStreamDeadlines(stream)
m, ok := msg.(*types.SSZUint64)
if !ok {
return fmt.Errorf("wrong message type for goodbye, got %T, wanted *uint64", msg)
}
if err := s.rateLimiter.validateRequest(stream, 1); err != nil {
return err
}
s.rateLimiter.add(stream, 1)
log := log.WithField("Reason", goodbyeMessage(*m))
log.WithField("peer", stream.Conn().RemotePeer()).Debug("Peer has sent a goodbye message")
s.p2p.Peers().SetNextValidTime(stream.Conn().RemotePeer(), goodByeBackoff(*m))
// closes all streams with the peer
return s.p2p.Disconnect(stream.Conn().RemotePeer())
}
// disconnectBadPeer checks whether peer is considered bad by some scorer, and tries to disconnect
// the peer, if that is the case. Additionally, disconnection reason is obtained from scorer.
func (s *Service) disconnectBadPeer(ctx context.Context, id peer.ID) {
if !s.p2p.Peers().IsBad(id) {
return
}
goodbyeCode := types.ErrToGoodbyeCode(s.p2p.Peers().Scorers().ValidationError(id))
if err := s.sendGoodByeAndDisconnect(ctx, goodbyeCode, id); err != nil {
log.Debugf("Error when disconnecting with bad peer: %v", err)
}
}
// A custom goodbye method that is used by our connection handler, in the
// event we receive bad peers.
func (s *Service) sendGoodbye(ctx context.Context, id peer.ID) error {
return s.sendGoodByeAndDisconnect(ctx, types.GoodbyeCodeGenericError, id)
}
func (s *Service) sendGoodByeAndDisconnect(ctx context.Context, code types.RPCGoodbyeCode, id peer.ID) error {
if err := s.sendGoodByeMessage(ctx, code, id); err != nil {
log.WithFields(logrus.Fields{
"error": err,
"peer": id,
}).Debug("Could not send goodbye message to peer")
}
return s.p2p.Disconnect(id)
}
func (s *Service) sendGoodByeMessage(ctx context.Context, code types.RPCGoodbyeCode, id peer.ID) error {
ctx, cancel := context.WithTimeout(ctx, respTimeout)
defer cancel()
stream, err := s.p2p.Send(ctx, &code, p2p.RPCGoodByeTopic, id)
if err != nil {
return err
}
defer closeStream(stream, log)
log := log.WithField("Reason", goodbyeMessage(code))
log.WithField("peer", stream.Conn().RemotePeer()).Debug("Sending Goodbye message to peer")
// Wait up to the response timeout for the peer to receive the goodbye
// and close the stream (or disconnect). We usually don't bother waiting
// around for an EOF, but we're going to close this connection
// immediately after we say goodbye.
//
// NOTE: we don't actually check the response as there's nothing we can
// do if something fails. We just need to wait for it.
SetStreamReadDeadline(stream, respTimeout)
_, _err := stream.Read([]byte{0})
_ = _err
return nil
}
func goodbyeMessage(num types.RPCGoodbyeCode) string {
reason, ok := types.GoodbyeCodeMessages[num]
if ok {
return reason
}
return fmt.Sprintf("unknown goodbye value of %d received", num)
}
// determines which backoff time to use depending on the
// goodbye code provided.
func goodByeBackoff(num types.RPCGoodbyeCode) time.Time {
duration, ok := backOffTime[num]
if !ok {
return time.Time{}
}
return time.Now().Add(duration)
}