p2p

Documentation

Overview

Package p2p implements the worker committee gossip network.

Index

• Constants
• Variables
• func DebugForceAllowUnroutableAddresses()
• func PublicKeyToPeerID(pk signature.PublicKey) (core.PeerID, error)
• type CommitteeMessage
• type Handler
• type ImportanceKind
  • func (ik ImportanceKind) Tag(runtimeID common.Namespace) string
  • func (ik ImportanceKind) TagValue() int
• type P2P
  • func New(ctx context.Context, identity *identity.Identity, consensus consensus.Backend) (*P2P, error)
  • func (p *P2P) Addresses() []node.Address
  • func (p *P2P) BlockPeer(peerID core.PeerID)
  • func (p *P2P) GetHost() core.Host
  • func (p *P2P) GetMinRepublishInterval() time.Duration
  • func (p *P2P) Peers(runtimeID common.Namespace) []string
  • func (p *P2P) PublishCommittee(ctx context.Context, runtimeID common.Namespace, msg *CommitteeMessage)
  • func (p *P2P) PublishTx(ctx context.Context, runtimeID common.Namespace, msg TxMessage)
  • func (p *P2P) RegisterHandler(runtimeID common.Namespace, kind TopicKind, handler Handler)
  • func (p *P2P) RegisterProtocolServer(srv rpc.Server)
• type PeerManager
  • func (mgr *PeerManager) Initialized() <-chan struct{}
  • func (mgr *PeerManager) KnownPeers() []core.PeerID
  • func (mgr *PeerManager) SetNodeImportance(kind ImportanceKind, runtimeID common.Namespace, ...)
  • func (mgr *PeerManager) SetNodes(nodes []*node.Node)
  • func (mgr *PeerManager) UpdateNode(node *node.Node) error
• type TopicKind
• type TxMessage

Constants

const (
	// 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"
	// CfgP2PValidateConcurrency sets the libp2p gossipsub per topic validator concurrency limit.
	// Note: this is a per-topic concurrency limit. We use one topic per runtime.
	CfgP2PValidateConcurrency = "worker.p2p.validate_concurrency"
	// CfgP2PValidateThrottle sets the libp2p gossipsub validator concurrency limit.
	// Note: this is a global (across all topics) validator concurrency limit.
	CfgP2PValidateThrottle = "worker.p2p.validate_throttle"
	// 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"
	// CfgP2PMaxNumPeers is the maximum number of peers.
	CfgP2PMaxNumPeers = "worker.p2p.max_num_peers"
	// CfgP2PPeerGracePeriod is the peer grace period.
	CfgP2PPeerGracePeriod = "worker.p2p.peer_grace_period"
)

const (
	ImportantNodeCompute    = 1
	ImportantNodeKeyManager = 2
)

Variables

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

Flags has the configuration flags.

Functions

func DebugForceAllowUnroutableAddresses

func DebugForceAllowUnroutableAddresses()

DebugForceAllowUnroutableAddresses allows unroutable addresses.

func PublicKeyToPeerID

func PublicKeyToPeerID(pk signature.PublicKey) (core.PeerID, error)

PublicKeyToPeerID converts a public key to a peer identifier.

Types

type CommitteeMessage

type CommitteeMessage struct {
	// Epoch is the epoch this message belongs to.
	Epoch beacon.EpochTime `json:"epoch,omitempty"`

	// Proposal is a batch proposal.
	Proposal *commitment.Proposal `json:",omitempty"`
}

CommitteeMessage is a message published to nodes via gossipsub on the committee topic.

type Handler

type Handler interface {
	// DecodeMessage decodes the given incoming message.
	DecodeMessage(msg []byte) (interface{}, error)

	// AuthorizeMessage handles authorizing an incoming message.
	//
	// The message handler will be re-invoked on error with a periodic backoff unless errors are
	// wrapped via `p2pError.Permanent`.
	AuthorizeMessage(ctx context.Context, peerID signature.PublicKey, msg interface{}) error

	// HandleMessage 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`.
	HandleMessage(ctx context.Context, peerID signature.PublicKey, msg interface{}, isOwn bool) error
}

Handler is a handler for P2P messages.

type ImportanceKind

type ImportanceKind uint8

ImportanceKind is the node importance kind.

func (ImportanceKind) Tag

func (ik ImportanceKind) Tag(runtimeID common.Namespace) string

Tag returns the connection manager tag associated with the given importance kind.

func (ImportanceKind) TagValue

func (ik ImportanceKind) TagValue() int

TagValue returns the connection manager tag value associated with the given importance kind.

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) BlockPeer

func (p *P2P) BlockPeer(peerID core.PeerID)

BlockPeer blocks a specific peer from being used by the local node.

func (*P2P) GetHost

func (p *P2P) GetHost() core.Host

GetHost returns the P2P host.

func (*P2P) GetMinRepublishInterval

func (p *P2P) GetMinRepublishInterval() time.Duration

GetMinRepublishInterval returns the minimum republish interval that needs to be respected by the caller when publishing the same message. If Publish is called for the same message more quickly, the message may be dropped and not published.

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) PublishCommittee

func (p *P2P) PublishCommittee(ctx context.Context, runtimeID common.Namespace, msg *CommitteeMessage)

PublishCommittee publishes a committee message.

func (*P2P) PublishTx

func (p *P2P) PublishTx(ctx context.Context, runtimeID common.Namespace, msg TxMessage)

PublishCommittee publishes a transaction message.

func (*P2P) RegisterHandler

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

RegisterHandler registers a message handler for the specified runtime and topic kind.

func (*P2P) RegisterProtocolServer

func (p *P2P) RegisterProtocolServer(srv rpc.Server)

RegisterProtocolServer registers a protocol server for the given protocol.

type PeerManager

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

PeerManager handles managing peers in the gossipsub network.

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) SetNodeImportance

func (mgr *PeerManager) SetNodeImportance(kind ImportanceKind, runtimeID common.Namespace, p2pIDs map[signature.PublicKey]bool)

SetNodeImportance configures node importance for the given set of nodes.

This makes it less likely for those nodes to be pruned.

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.

type TopicKind

type TopicKind string

TopicKind is the gossipsub topic kind.

const (
	// TopicKindCommittee is the topic kind for the topic that is used to gossip batch proposals
	// and other committee messages.
	TopicKindCommittee TopicKind = "committee"
	// TopicKindTx is the topic kind for the topic that is used to gossip transactions.
	TopicKindTx TopicKind = "tx"
)

type TxMessage

type TxMessage []byte

TxMessage is a message published to nodes via gossipsub on the transaction topic. It contains the raw signed transaction with runtime-dependent semantics.