p2p

Documentation

Overview

Package p2p implements the worker committee gossip network.

Index

• Constants
• Variables
• func DebugForceAllowUnroutableAddresses()
• func Enabled() bool
• type BaseHandler
  • func (h *BaseHandler) AuthenticatePeer(peerID signature.PublicKey, msg *Message) error
  • func (h *BaseHandler) HandlePeerMessage(peerID signature.PublicKey, msg *Message, isOwn bool) error
• type Handler
• type Message
• type P2P
  • func New(ctx context.Context, identity *identity.Identity, consensus consensus.Backend) (*P2P, error)
  • func (p *P2P) Addresses() []node.Address
  • func (p *P2P) Peers(runtimeID common.Namespace) []string
  • func (p *P2P) Publish(ctx context.Context, runtimeID common.Namespace, msg *Message)
  • func (p *P2P) RegisterHandler(runtimeID common.Namespace, handler Handler)
• type PeerManager
  • func (mgr *PeerManager) Initialized() <-chan struct{}
  • func (mgr *PeerManager) KnownPeers() []core.PeerID
  • func (mgr *PeerManager) SetNodes(nodes []*node.Node)
  • func (mgr *PeerManager) UpdateNode(node *node.Node) error

Constants

const (
	// CfgP2PEnabled enables the P2P worker (automatically enabled if compute worker enabled).
	CfgP2PEnabled = "worker.p2p.enabled"

	// CfgP2pPort configures the P2P port.
	CfgP2pPort = "worker.p2p.port"

	// CfgP2PPeerOutboundQueueSize sets the libp2p gossipsub buffer size for outbound messages.
	CfgP2PPeerOutboundQueueSize = "worker.p2p.peer_outbound_queue_size"
	// CfgP2PValidateQueueSize sets the libp2p gossipsub buffer size of the validate queue.
	CfgP2PValidateQueueSize = "worker.p2p.validate_queue_size"
	// CfgP2PConnectednessLowWater sets the ratio of connected to unconnected peers at which
	// the peer manager will try to reconnect to disconnected nodes.
	CfgP2PConnectednessLowWater = "worker.p2p.connectedness_low_water"
)

Variables

var Flags = flag.NewFlagSet("", flag.ContinueOnError)

Flags has the configuration flags.

Functions

func DebugForceAllowUnroutableAddresses

func DebugForceAllowUnroutableAddresses()

DebugForceAllowUnroutableAddresses allows unroutable addresses.

func Enabled

func Enabled() bool

Enabled reads our enabled flag from viper.

Types

type BaseHandler

type BaseHandler struct {
}

BaseHandler handler is a P2P handler that can be used in publishing-only clients.

func (*BaseHandler) AuthenticatePeer

func (h *BaseHandler) AuthenticatePeer(peerID signature.PublicKey, msg *Message) error

AuthenticatePeer implements p2p Handler.

func (*BaseHandler) HandlePeerMessage

func (h *BaseHandler) HandlePeerMessage(peerID signature.PublicKey, msg *Message, isOwn bool) error

HandlePeerMessage implements p2p Handler.

type Handler

type Handler interface {
	// AuthenticatePeer handles authenticating a peer that send an
	// incoming message.
	//
	// The message handler will be re-invoked on error with a periodic
	// backoff unless errors are wrapped via `p2pError.Permanent`.
	AuthenticatePeer(peerID signature.PublicKey, msg *Message) error

	// HandlePeerMessage handles an incoming message from a peer.
	//
	// The message handler will be re-invoked on error with a periodic
	// backoff unless errors are wrapped via `p2pError.Permanent`.
	HandlePeerMessage(peerID signature.PublicKey, msg *Message, isOwn bool) error
}

Handler is a handler for P2P messages.

type Message

type Message struct {
	// GroupVersion is the version of all elected committees (the consensus
	// block height of last processed committee election). Messages with
	// non-matching group versions will be discarded.
	GroupVersion int64 `json:"group_version,omitempty"`

	// Jaeger's span context in binary format.
	SpanContext []byte `json:"span,omitempty"`

	ProposedBatch  *commitment.SignedProposedBatch `json:",omitempty"`
	ExecutorCommit *commitment.ExecutorCommitment  `json:",omitempty"`
	Tx             *executor.Tx                    `json:",omitempty"`
}

Message is a message sent to nodes via P2P transport.

type P2P

type P2P struct {
	sync.RWMutex
	*PeerManager
	// contains filtered or unexported fields
}

P2P is a peer-to-peer node using libp2p.

func New

func New(ctx context.Context, identity *identity.Identity, consensus consensus.Backend) (*P2P, error)

New creates a new P2P node.

func (*P2P) Addresses

func (p *P2P) Addresses() []node.Address

Addresses returns the P2P addresses of the node.

func (*P2P) Peers

func (p *P2P) Peers(runtimeID common.Namespace) []string

Peers returns a list of connected P2P peers for the given runtime.

func (*P2P) Publish

func (p *P2P) Publish(ctx context.Context, runtimeID common.Namespace, msg *Message)

Publish publishes a message to the gossip network.

func (*P2P) RegisterHandler

func (p *P2P) RegisterHandler(runtimeID common.Namespace, handler Handler)

RegisterHandler registers a message handler for the specified runtime. If multiple handlers are registered for the same runtime, each of the handlers will get invoked.

type PeerManager

type PeerManager struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

PeerManager handles managing peers in the gossipsub network.

XXX: we accept connections from all peers, however known peers from registry are considered trustworthier and we maintain persistent connections with them. Once libp2p layer supports "peer reputation" configure better reputation for registry peers.

func (*PeerManager) Initialized

func (mgr *PeerManager) Initialized() <-chan struct{}

Initialized returns a channel that will be closed once the manager is initialized and has received the first node refresh event.

func (*PeerManager) KnownPeers

func (mgr *PeerManager) KnownPeers() []core.PeerID

KnownPeers returns a list of currently known peer IDs.

func (*PeerManager) SetNodes

func (mgr *PeerManager) SetNodes(nodes []*node.Node)

SetNodes sets the membership of the gossipsub network.

func (*PeerManager) UpdateNode

func (mgr *PeerManager) UpdateNode(node *node.Node) error

UpdateNode upserts a node into the gossipsub network.