p2p

README

drtg-dharitri p2p components

The Messenger interface with its implementation are used to define the way to communicate between Dharitri nodes.

There are 2 ways to send data to the other peers:

1. Broadcasting messages on a pubsub using topics;
2. Direct sending messages to the connected peers.

The first type is used to send messages that have to reach every node (from corresponding shard, metachain, consensus group, etc.). The second type is used to resolve requests coming from directly connected peers.

Documentation

Index

• Constants
• Variables
• func MessageOriginatorPid(msg MessageP2P) string
• func MessageOriginatorSeq(msg MessageP2P) string
• func PeerIdToShortString(pid core.PeerID) string
• type BroadcastMethod
• type ConnectedPeersInfo
• type ConnectionsHandler
• type ConnectionsWatcher
• type Debugger
• type DirectSender
• type Logger
• type Marshaller
• type MessageHandler
• type MessageP2P
• type MessageProcessor
• type Messenger
• type NetworkShardingCollector
• type NetworkType
• type P2PKeyConverter
• type PeerCounts
• type PeerDenialEvaluator
• type PeerDiscoverer
• type PeerDiscoveryFactory
• type PeerShardResolver
• type PeerTopicNotifier
• type PeersRatingHandler
• type PeersRatingMonitor
• type PreferredPeersHolderHandler
• type Reconnecter
• type Sharder
• type SignerVerifier
• type SyncTimer

Constants

const (
	// LocalHostListenAddrWithIp4AndTcp defines the local host listening ip v.4 address and TCP
	LocalHostListenAddrWithIp4AndTcp = "/ip4/127.0.0.1/tcp/%d"

	// ListsSharder is the variant that uses lists
	ListsSharder = "ListsSharder"
	// OneListSharder is the variant that is shard agnostic and uses one list
	OneListSharder = "OneListSharder"
	// NilListSharder is the variant that will not do connection trimming
	NilListSharder = "NilListSharder"

	// ConnectionWatcherTypePrint - new connection found will be printed in the log file
	ConnectionWatcherTypePrint = "print"
	// ConnectionWatcherTypeDisabled - no connection watching should be made
	ConnectionWatcherTypeDisabled = "disabled"
	// ConnectionWatcherTypeEmpty - not set, no connection watching should be made
	ConnectionWatcherTypeEmpty = ""

	// WrongP2PMessageBlacklistDuration represents the time to keep a peer id in the blacklist if it sends a message that
	// do not follow this protocol
	WrongP2PMessageBlacklistDuration = time.Second * 7200
)

Variables

var ErrAlreadySeenMessage = errors.New("already seen this message")

ErrAlreadySeenMessage signals that the message has already been seen

var ErrChannelCanNotBeDeleted = errors.New("channel can not be deleted")

ErrChannelCanNotBeDeleted signals that a channel can not be deleted (might be the default channel)

var ErrChannelCanNotBeReAdded = errors.New("channel can not be re added")

ErrChannelCanNotBeReAdded signals that a channel can not be re added as it is the default channel

var ErrChannelDoesNotExist = errors.New("channel does not exist")

ErrChannelDoesNotExist signals that a requested channel does not exist

var ErrEmptyBufferToSend = errors.New("empty buffer to send")

ErrEmptyBufferToSend signals that an empty buffer was provided for sending to other peers

var ErrEndPortIsSmallerThanStartPort = errors.New("ending port value is smaller than the starting port value")

ErrEndPortIsSmallerThanStartPort signals that the ending port value is smaller than the starting port value

var ErrInvalidConfig = errors.New("invalid config")

ErrInvalidConfig signals that an invalid config has been provided

var ErrInvalidDurationProvided = errors.New("invalid time.Duration provided")

ErrInvalidDurationProvided signals that an invalid time.Duration has been provided

var ErrInvalidEndingPortValue = errors.New("invalid ending port value")

ErrInvalidEndingPortValue signals that an invalid ending port value has been provided

var ErrInvalidPortValue = errors.New("invalid port value")

ErrInvalidPortValue signals that an invalid port value has been provided

var ErrInvalidPortsRangeString = errors.New("invalid ports range string")

ErrInvalidPortsRangeString signals that an invalid ports range string has been provided

var ErrInvalidQUICAddress = errors.New("invalid QUIC address")

ErrInvalidQUICAddress signals that an invalid QUIC address was used

var ErrInvalidSeedersReconnectionInterval = errors.New("invalid seeders reconnection interval")

ErrInvalidSeedersReconnectionInterval signals that an invalid seeders reconnection interval error occurred

var ErrInvalidStartingPortValue = errors.New("invalid starting port value")

ErrInvalidStartingPortValue signals that an invalid starting port value has been provided

var ErrInvalidTCPAddress = errors.New("invalid TCP address")

ErrInvalidTCPAddress signals that an invalid TCP address was used

var ErrInvalidValue = errors.New("invalid value")

ErrInvalidValue signals that an invalid value has been provided

var ErrInvalidWSAddress = errors.New("invalid WebSocket address")

ErrInvalidWSAddress signals that an invalid WebSocket address was used

var ErrInvalidWebTransportAddress = errors.New("invalid WebTransport address")

ErrInvalidWebTransportAddress signals that an invalid WebTransport address was used

var ErrMessageProcessorAlreadyDefined = errors.New("message processor already defined")

ErrMessageProcessorAlreadyDefined signals that a message processor was already defined on the provided topic and identifier

var ErrMessageProcessorDoesNotExists = errors.New("message processor does not exists")

ErrMessageProcessorDoesNotExists signals that a message processor does not exist on the provided topic and identifier

var ErrMessageTooLarge = errors.New("buffer too large")

ErrMessageTooLarge signals that the message provided is too large

var ErrMessageTooNew = errors.New("message is too new")

ErrMessageTooNew signals that a message has a timestamp that is in the future relative to self

var ErrMessageTooOld = errors.New("message is too old")

ErrMessageTooOld signals that a message has a timestamp that is in the past relative to self

var ErrMessageUnmarshalError = errors.New("message unmarshal error")

ErrMessageUnmarshalError signals that an invalid message was received from a peer. There is no way to communicate with such a peer as it does not respect the protocol

var ErrNilCacher = errors.New("nil cacher")

ErrNilCacher signals that a nil cacher has been provided

var ErrNilChannelLoadBalancer = errors.New("nil channel load balancer")

ErrNilChannelLoadBalancer signals that a nil channel load balancer has been provided

var ErrNilConnectionMonitor = errors.New("nil connections monitor")

ErrNilConnectionMonitor signals that a nil connections monitor has been provided

var ErrNilConnectionsHandler = errors.New("nil connections handler")

ErrNilConnectionsHandler signals that a nil connections handler has been provided

var ErrNilConnectionsMetric = errors.New("nil connections metric")

ErrNilConnectionsMetric signals that a nil connections metric has been provided

var ErrNilConnectionsWatcher = errors.New("nil connections watcher")

ErrNilConnectionsWatcher signals that a nil connections watcher has been provided

var ErrNilContext = errors.New("nil context")

ErrNilContext signals that a nil context was provided

var ErrNilDebugger = errors.New("nil debugger")

ErrNilDebugger signals that a nil debugger has been provided

var ErrNilDirectSendMessageHandler = errors.New("nil direct sender message handler")

ErrNilDirectSendMessageHandler signals that the message handler for new message has not been wired

var ErrNilDirectSender = errors.New("nil direct sender")

ErrNilDirectSender signals that a nil direct sender has been provided

var ErrNilFetchPeersOnTopicHandler = errors.New("nil fetch peers on topic handler")

ErrNilFetchPeersOnTopicHandler signals that a nil handler was provided

var ErrNilHost = errors.New("nil host")

ErrNilHost signals that a nil host has been provided

var ErrNilLogger = errors.New("nil logger")

ErrNilLogger signals that a nil logger has been provided

var ErrNilMarshaller = errors.New("nil marshaller")

ErrNilMarshaller signals that an operation has been attempted to or with a nil marshaller implementation

var ErrNilMessage = errors.New("nil message")

ErrNilMessage signals that a nil message has been received

var ErrNilMockNet = errors.New("nil mocknet provided")

ErrNilMockNet signals that a nil mocknet was provided. Should occur only in testing!!!

var ErrNilNetwork = errors.New("nil network")

ErrNilNetwork signals that a nil network has been provided

var ErrNilP2PSigner = errors.New("nil p2p signer")

ErrNilP2PSigner signals that a nil p2p signer has been provided

var ErrNilP2pKeyGenerator = errors.New("nil p2p key generator")

ErrNilP2pKeyGenerator signals that a nil p2p key generator has been provided

var ErrNilP2pPrivateKey = errors.New("nil p2p private key")

ErrNilP2pPrivateKey signals that a nil p2p private key has been provided

var ErrNilP2pSingleSigner = errors.New("nil p2p single signer")

ErrNilP2pSingleSigner signals that a nil p2p single signer has been provided

var ErrNilPeerDenialEvaluator = errors.New("nil peer denial evaluator")

ErrNilPeerDenialEvaluator signals that a nil peer denial evaluator was provided

var ErrNilPeerDiscoverer = errors.New("nil peer discoverer")

ErrNilPeerDiscoverer signals that a nil peer discoverer has been provided

var ErrNilPeerShardResolver = errors.New("nil PeerShardResolver")

ErrNilPeerShardResolver signals that the peer shard resolver provided is nil

var ErrNilPeerTopicNotifier = errors.New("nil peer topic notifier")

ErrNilPeerTopicNotifier signals that a nil peer topic notifier have been provided

var ErrNilPeersOnChannel = errors.New("nil peers on channel")

ErrNilPeersOnChannel signals that a nil peers on channel has been provided

var ErrNilPeersRatingHandler = errors.New("nil peers rating handler")

ErrNilPeersRatingHandler signals that a nil peers rating handler has been provided

var ErrNilPreferredPeersHolder = errors.New("nil peers holder")

ErrNilPreferredPeersHolder signals that a nil preferred peers holder was provided

var ErrNilPubSub = errors.New("nil pubSub")

ErrNilPubSub signals that a nil pubSub has been provided

var ErrNilReconnecter = errors.New("nil reconnecter")

ErrNilReconnecter signals that a nil reconnecter has been provided

var ErrNilSharder = errors.New("nil sharder")

ErrNilSharder signals that the provided sharder is nil

var ErrNilSyncTimer = errors.New("nil sync timer")

ErrNilSyncTimer signals that a nil sync timer was provided

var ErrNilThrottler = errors.New("nil throttler")

ErrNilThrottler signals that a nil throttler has been provided

var ErrNilTopic = errors.New("nil topic")

ErrNilTopic signals that a nil topic has been provided

var ErrNilValidator = errors.New("no validator has been set for this topic")

ErrNilValidator signals that a validator hasn't been set for the required topic

var ErrNoFreePortInRange = errors.New("no free port in range")

ErrNoFreePortInRange signals that no free port was found from provided range

var ErrNoTransportsDefined = errors.New("no transports defined")

ErrNoTransportsDefined signals that no transports were defined

var ErrPeerDiscoveryProcessAlreadyStarted = errors.New("peer discovery is already turned on")

ErrPeerDiscoveryProcessAlreadyStarted signals that a peer discovery is already turned on

var ErrPeerNotDirectlyConnected = errors.New("peer is not directly connected")

ErrPeerNotDirectlyConnected signals that the peer is not directly connected to self

var ErrTooManyGoroutines = errors.New(" number of goroutines exceeded")

ErrTooManyGoroutines is raised when the number of goroutines has exceeded a threshold

var ErrUnsupportedFields = errors.New("unsupported fields")

ErrUnsupportedFields signals that unsupported fields are provided

var ErrUnsupportedMessageVersion = errors.New("unsupported message version")

ErrUnsupportedMessageVersion signals that an unsupported message version was detected

var ErrUnwantedPeer = errors.New("unwanted peer: will not initiate connection as it will get disconnected")

ErrUnwantedPeer signals that the provided peer has a longer kademlia distance in respect with the already connected peers and a connection to this peer will result in an immediate disconnection

var ErrWrongTypeAssertion = errors.New("wrong type assertion")

ErrWrongTypeAssertion signals that a wrong type assertion occurred

var ErrWrongTypeAssertions = errors.New("wrong type assertion")

ErrWrongTypeAssertions signals that a wrong type assertion occurred

Functions

func MessageOriginatorPid

func MessageOriginatorPid(msg MessageP2P) string

MessageOriginatorPid will output the message peer id in a pretty format If it can, it will display the last displayLastPidChars (12) characters from the pid

func MessageOriginatorSeq

func MessageOriginatorSeq(msg MessageP2P) string

MessageOriginatorSeq will output the sequence number as hex

func PeerIdToShortString

func PeerIdToShortString(pid core.PeerID) string

PeerIdToShortString trims the first displayLastPidChars characters of the provided peer ID after converting the peer ID to string using the Pretty functionality

Types

type BroadcastMethod

type BroadcastMethod string

BroadcastMethod defines the broadcast method of the message

const Broadcast BroadcastMethod = "Broadcast"

Broadcast defines a broadcast message

const Direct BroadcastMethod = "Direct"

Direct defines a direct message

type ConnectedPeersInfo

type ConnectedPeersInfo struct {
	SelfShardID              uint32
	UnknownPeers             []string
	Seeders                  []string
	IntraShardValidators     map[uint32][]string
	IntraShardObservers      map[uint32][]string
	CrossShardValidators     map[uint32][]string
	CrossShardObservers      map[uint32][]string
	NumValidatorsOnShard     map[uint32]int
	NumObserversOnShard      map[uint32]int
	NumPreferredPeersOnShard map[uint32]int
	NumIntraShardValidators  int
	NumIntraShardObservers   int
	NumCrossShardValidators  int
	NumCrossShardObservers   int
}

ConnectedPeersInfo represents the DTO structure used to output the metrics for connected peers

type ConnectionsHandler

type ConnectionsHandler interface {
	io.Closer

	Bootstrap() error
	Peers() []core.PeerID
	Addresses() []string
	ConnectToPeer(address string) error
	IsConnected(peerID core.PeerID) bool
	ConnectedPeers() []core.PeerID
	ConnectedAddresses() []string
	PeerAddresses(pid core.PeerID) []string
	ConnectedPeersOnTopic(topic string) []core.PeerID
	SetPeerShardResolver(peerShardResolver PeerShardResolver) error
	GetConnectedPeersInfo() *ConnectedPeersInfo
	WaitForConnections(maxWaitingTime time.Duration, minNumOfPeers uint32)
	IsConnectedToTheNetwork() bool
	ThresholdMinConnectedPeers() int
	SetThresholdMinConnectedPeers(minConnectedPeers int) error
	SetPeerDenialEvaluator(handler PeerDenialEvaluator) error
	IsInterfaceNil() bool
}

ConnectionsHandler defines the behaviour of a component able to handle connections

type ConnectionsWatcher

type ConnectionsWatcher interface {
	NewKnownConnection(pid core.PeerID, connection string)
	Close() error
	IsInterfaceNil() bool
}

ConnectionsWatcher represent an entity able to watch new connections

type Debugger

type Debugger interface {
	AddIncomingMessage(topic string, size uint64, isRejected bool)
	AddOutgoingMessage(topic string, size uint64, isRejected bool)
	Close() error
	IsInterfaceNil() bool
}

Debugger represent a p2p debugger able to print p2p statistics (messages received/sent per topic)

type DirectSender

type DirectSender interface {
	NextSequenceNumber() []byte
	Send(topic string, buff []byte, peer core.PeerID) error
	RegisterDirectMessageProcessor(handler MessageHandler) error
	IsInterfaceNil() bool
}

DirectSender defines a component that can send direct messages to connected peers

type Logger

type Logger interface {
	Trace(message string, args ...interface{})
	Debug(message string, args ...interface{})
	Info(message string, args ...interface{})
	Warn(message string, args ...interface{})
	Error(message string, args ...interface{})
	LogIfError(err error, args ...interface{})
	GetLevel() logger.LogLevel
	IsInterfaceNil() bool
}

Logger defines the behavior of a data logger component

type Marshaller

type Marshaller interface {
	Marshal(obj interface{}) ([]byte, error)
	Unmarshal(obj interface{}, buff []byte) error
	IsInterfaceNil() bool
}

Marshaller defines the 2 basic operations: serialize (marshal) and deserialize (unmarshal)

type MessageHandler

type MessageHandler interface {
	io.Closer
	MessageProcessor

	CreateTopic(name string, createChannelForTopic bool) error
	HasTopic(name string) bool
	RegisterMessageProcessor(topic string, identifier string, handler MessageProcessor) error
	UnregisterAllMessageProcessors() error
	UnregisterMessageProcessor(topic string, identifier string) error
	Broadcast(topic string, buff []byte)
	BroadcastOnChannel(channel string, topic string, buff []byte)
	BroadcastUsingPrivateKey(topic string, buff []byte, pid core.PeerID, skBytes []byte)
	BroadcastOnChannelUsingPrivateKey(channel string, topic string, buff []byte, pid core.PeerID, skBytes []byte)
	SendToConnectedPeer(topic string, buff []byte, peerID core.PeerID) error
	UnJoinAllTopics() error
	IsInterfaceNil() bool
}

MessageHandler defines the behaviour of a component able to send and process messages

type MessageP2P

type MessageP2P interface {
	From() []byte
	Data() []byte
	Payload() []byte
	SeqNo() []byte
	Topic() string
	Signature() []byte
	Key() []byte
	Peer() core.PeerID
	Timestamp() int64
	BroadcastMethod() BroadcastMethod
	IsInterfaceNil() bool
}

MessageP2P defines what a p2p message can do (should return)

type MessageProcessor

type MessageProcessor interface {
	ProcessReceivedMessage(message MessageP2P, fromConnectedPeer core.PeerID, source MessageHandler) error
	IsInterfaceNil() bool
}

MessageProcessor is the interface used to describe what a receive message processor should do All implementations that will be called from Messenger implementation will need to satisfy this interface If the function returns a non nil value, the received message will not be propagated to its connected peers

type Messenger

type Messenger interface {
	MessageHandler
	ConnectionsHandler

	ID() core.PeerID

	Port() int
	Sign(payload []byte) ([]byte, error)
	Verify(payload []byte, pid core.PeerID, signature []byte) error
	SignUsingPrivateKey(skBytes []byte, payload []byte) ([]byte, error)
	AddPeerTopicNotifier(notifier PeerTopicNotifier) error
	IsInterfaceNil() bool
}

Messenger is the main interface used for communication with other peers

type NetworkShardingCollector

type NetworkShardingCollector interface {
	UpdatePeerIDInfo(pid core.PeerID, pk []byte, shardID uint32)
	IsInterfaceNil() bool
}

NetworkShardingCollector defines the updating methods used by the network sharding component The interface assures that the collected data will be used by the p2p network sharding components

type NetworkType

type NetworkType string

NetworkType defines the type of the network a messenger is running on

type P2PKeyConverter

type P2PKeyConverter interface {
	ConvertPeerIDToPublicKey(keyGen crypto.KeyGenerator, pid core.PeerID) (crypto.PublicKey, error)
	ConvertPublicKeyToPeerID(pk crypto.PublicKey) (core.PeerID, error)
	IsInterfaceNil() bool
}

P2PKeyConverter defines what a p2p key converter can do

type PeerCounts

type PeerCounts struct {
	UnknownPeers    int
	IntraShardPeers int
	CrossShardPeers int
}

PeerCounts represents the DTO structure used to output the count metrics for connected peers

type PeerDenialEvaluator

type PeerDenialEvaluator interface {
	IsDenied(pid core.PeerID) bool
	UpsertPeerID(pid core.PeerID, duration time.Duration) error
	IsInterfaceNil() bool
}

PeerDenialEvaluator defines the behavior of a component that is able to decide if a peer ID is black listed or not TODO move antiflooding inside network messenger

type PeerDiscoverer

type PeerDiscoverer interface {
	Bootstrap() error
	Name() string
	IsInterfaceNil() bool
}

PeerDiscoverer defines the behaviour of a peer discovery mechanism

type PeerDiscoveryFactory

type PeerDiscoveryFactory interface {
	CreatePeerDiscoverer() (PeerDiscoverer, error)
	IsInterfaceNil() bool
}

PeerDiscoveryFactory defines the factory for peer discoverer implementation

type PeerShardResolver

type PeerShardResolver interface {
	GetPeerInfo(pid core.PeerID) core.P2PPeerInfo
	IsInterfaceNil() bool
}

PeerShardResolver is able to resolve the link between the provided PeerID and the shardID

type PeerTopicNotifier

type PeerTopicNotifier interface {
	NewPeerFound(pid core.PeerID, topic string)
	IsInterfaceNil() bool
}

PeerTopicNotifier represent an entity able to handle new notifications on a new peer on a topic

type PeersRatingHandler

type PeersRatingHandler interface {
	IncreaseRating(pid core.PeerID)
	DecreaseRating(pid core.PeerID)
	GetTopRatedPeersFromList(peers []core.PeerID, minNumOfPeersExpected int) []core.PeerID
	IsInterfaceNil() bool
}

PeersRatingHandler represent an entity able to handle peers ratings

type PeersRatingMonitor

type PeersRatingMonitor interface {
	GetConnectedPeersRatings(connectionsHandler ConnectionsHandler) (string, error)
	IsInterfaceNil() bool
}

PeersRatingMonitor represent an entity able to provide peers ratings

type PreferredPeersHolderHandler

type PreferredPeersHolderHandler interface {
	PutConnectionAddress(peerID core.PeerID, address string)
	PutShardID(peerID core.PeerID, shardID uint32)
	Get() map[uint32][]core.PeerID
	Contains(peerID core.PeerID) bool
	Remove(peerID core.PeerID)
	Clear()
	IsInterfaceNil() bool
}

PreferredPeersHolderHandler defines the behavior of a component able to handle preferred peers operations

type Reconnecter

type Reconnecter interface {
	ReconnectToNetwork(ctx context.Context)
	IsInterfaceNil() bool
}

Reconnecter defines the behaviour of a network reconnection mechanism

type Sharder

type Sharder interface {
	SetSeeders(addresses []string)
	IsSeeder(pid core.PeerID) bool
	SetPeerShardResolver(psp PeerShardResolver) error
	IsInterfaceNil() bool
}

Sharder defines the eviction computing process of unwanted peers

type SignerVerifier

type SignerVerifier interface {
	Sign(payload []byte) ([]byte, error)
	Verify(payload []byte, pid core.PeerID, signature []byte) error
	IsInterfaceNil() bool
}

SignerVerifier is used in higher level protocol authentication of 2 peers after the basic p2p connection has been made

type SyncTimer

type SyncTimer interface {
	CurrentTime() time.Time
	IsInterfaceNil() bool
}

SyncTimer represent an entity able to tell the current time