p2p

README

P2P

This package provides an interface for peer-to-peer networking features, such as peer identification, addressing, connectivity, messaging, custom protocol streams, and peer discovery. These features are implemented using libp2p, and this package and its sub-packages serve as the main integration point with libp2p.

The code in this package is part of an ongoing initiative to research and implement an alternative networking stack that supports decentralized network topologies, while benefiting from the history and experience of libp2p's protocol development.

The current status of this initiative is purely experimental, and is not enabled by default or recommended for regular use. The components implemented here offer no guarantee of security, reliability, or performance. Updates will be provided on the progress and readiness of these components for standard usage in the future.

Background

Decentralized networks depend on mechanisms for discovering other peers and reliably broadcasting messages to other interested peers. Libp2p provides pluggable transports and multiplexing to allow multiple stream-based protocols to share the same connection. Libp2p also provides an implementation of a message-based gossip protocol, GossipSub.

Algorand's current network protocol sends messages between peers over bidirectional WebSocket connections. Nodes that are configured to enable message-forwarding (including nodes currently called "relays") validate incoming messages, then selectively forward messages to other connected peers. This network implementation (WebsocketNetwork) sits behind the GossipNode interface in the network package.

Current P2P implementation details

This package supports a P2PNetwork implementation of the GossipNode interface, providing the same peer connection management and message broadcast functions, but via peer connections managed by libp2p. The P2PNetwork implementation uses libp2p's multiaddress scheme and peer IDs to establish connections and identify peers.

Currently transactions (protocol tag TX) are distributed using the GossipSub protocol (see pubsub.go), while all other messages are forwarded over the pre-existing custom message protocol /algorand-ws/1.0.0 (see streams.go) that uses the same message serialization as the existing WebsocketNetwork implementation. These two protocols are multiplexed over a single connection using libp2p streams.

graph LR
    subgraph "P2P Implementation"
        P2P[P2PNetwork]
        TX[Transactions]
        AW[Other messages]
    end
    P2P --> TX
    P2P --> AW

    subgraph "libp2p"
        G["GossipSub 1.1<br/>/meshsub/1.1.0"]
        WS["Legacy message serialization<br/>/algorand-ws/1.0.0"]
        S["Multiplexer<br/>/yamux/1.0.0"]
        T["TCP Transport"]
    end
    G --> S
    WS --> S
    TX --> G
    AW --> WS
    S --> T

The underlying libp2p implementation is abstracted as p2p.Service and is initialized in two steps:

1. Creating a p2p Host
2. Creating a service serviceImpl object

Host is also used for p2p HTTP server and DHT Discovery service creation. It is also useful for unit testing. Note, Host is created with NoListenAddrs options that prevents automatic listening and networking until the Service.Start() is called. This follows the designs of Algod services (including the WsNetwork service).

Connection limiting

libp2p's ResourceManager is used to limit the number of connections up to cfg.IncomingConnectionsLimit.

DHT and capabilities

Provides helper methods to construct DHT discovery service using go-libp2p-kad-dht library. High level CapabilitiesDiscovery class supports retrieving (PeersForCapability) peers by a given capability(-ies) or advertising own capabilities (AdvertiseCapabilities).

Note, by default private and non-routable addresses are filtered (see AddrsFactory), libp2p's ObservedAddrManager can track its own public address and makes it available (and so that discoverable with DHT) if it was observed at least 4 times in 30 minutes (as of libp2p@v0.33.2).

graph LR

    subgraph "node"
        Cap[Capabilities]
    end

    subgraph "P2P Implementation"
        P2P[P2PNetwork]
        AdvCap[AdvertiseCapabilities]
    end

    P2P --> AdvCap
    Cap -.-> P2P

    subgraph "libp2p"
        Adv[Advertise]
        Addr[Addrs]
        OAM[ObservedAddrManager]
        AF[AddrFactory]
        KAD["/kad/1.0.0"]
    end

    OAM -.-> Addr
    AF -.-> Addr
    AdvCap --> Adv

    subgraph "libp2p-kad-dht"
        Pro[Provide]
    end

    Addr -.-> Pro
    Adv --> Pro
    Pro --> KAD

HTTP over libp2p connection

libp2p@0.33 added ability to multiplex HTTP traffic in p2p connection. A custom /algorand-http/1.0.0 stream is utilized to expose HTTP server and allow network service clients (catchup, catchpoint, txsync) to register its own handlers similarly to the legacy ws-net implementation.

Peerstore

In-memory peerstore implements libp2p.Peerstore and go-algorand Phonebook interfaces. Peer classes (relays, archival, etc) and persistent peers (i.e. peers from command line or phonebook.json) are supported. Possible enhancement is to save/load peerstore to/from disk to tolerate bootstrap nodes failures.

Logging

lip2p uses zap logger as a separate ipfs/go-log/v2 module. EnableP2PLogging helper adds go-algorand's logrus as a custom zap core so that all libp2p logs go through go-algorand logging facility. Unfortunately ipfs/go-log/v2 has a primary logging core as module variable that makes impossible to have custom logrus sub-loggers in unit tests.

Metrics

go-libp2p uses Prometheus as a metrics library, go-libp2p-kad-dht relies on OpenCensus library. go-algorand has two collectors (see util/metrics) for both Prometheus and OpenCensus for counters and gauges with labels. Other types (summary, histogram, distribution) are not supported at the moment.

Documentation

Index

• Constants
• Variables
• func EnableP2PLogging(log logging.Logger, l logging.Level) error
• func GetPeerTelemetryInfo(peerProtocols []protocol.ID) (telemetryID string, telemetryInstance string)
• func GetPrivKey(cfg config.Local, dataDir string) (crypto.PrivKey, error)
• func MakeHost(cfg config.Local, datadir string, pstore *pstore.PeerStore) (host.Host, string, error)
• func MakeService(ctx context.Context, log logging.Logger, cfg config.Local, h host.Host, ...) (*serviceImpl, error)
• func MakeTestHTTPClient(addrInfo *peer.AddrInfo, opts ...httpClientOption) (*http.Client, error)
• func SetP2PLogLevel(l logging.Level) error
• func WithHost(h host.Host) httpClientOption
• type CapabilitiesDiscovery
  • func MakeCapabilitiesDiscovery(ctx context.Context, cfg config.Local, h host.Host, ...) (*CapabilitiesDiscovery, error)
  • func (c *CapabilitiesDiscovery) AdvertiseCapabilities(capabilities ...Capability)
  • func (c *CapabilitiesDiscovery) Close() error
  • func (c *CapabilitiesDiscovery) Host() host.Host
  • func (c *CapabilitiesDiscovery) PeersForCapability(capability Capability, n int) ([]peer.AddrInfo, error)
  • func (c *CapabilitiesDiscovery) RoutingTable() Sizer
• type Capability
• type HTTPServer
  • func MakeHTTPServer(streamHost host.Host) *HTTPServer
  • func (s *HTTPServer) RegisterHTTPHandler(path string, handler http.Handler)
  • func (s *HTTPServer) RegisterHTTPHandlerFunc(path string, handler func(http.ResponseWriter, *http.Request))
• type PeerID
  • func PeerIDFromPublicKey(pubKey crypto.PubKey) (PeerID, error)
  • func (id PeerID) String() string
• type PeerIDChallengeSigner
  • func (p *PeerIDChallengeSigner) PublicKey() algocrypto.PublicKey
  • func (p *PeerIDChallengeSigner) Sign(message algocrypto.Hashable) algocrypto.Signature
  • func (p *PeerIDChallengeSigner) SignBytes(message []byte) algocrypto.Signature
• type Service
• type Sizer
• type StreamHandler
• type SubNextCancellable

Constants

const (
	// Archival nodes
	Archival Capability = "archival"
	// Catchpoints storing nodes
	Catchpoints = "catchpointStoring"
	// Gossip nodes are non permissioned relays
	Gossip = "gossip"
)

const AlgorandWsProtocol = "/algorand-ws/1.0.0"

AlgorandWsProtocol defines a libp2p protocol name for algorand's websockets messages

const DefaultPrivKeyPath = "peerIDPrivKey.key"

DefaultPrivKeyPath is the default path inside the node's root directory at which the private key for p2p identity is found and persisted to when a new one is generated.

const TXTopicName = "algotx01"

TXTopicName defines a pubsub topic for TX messages There is a micro optimization for const string comparison: 8 bytes const string require a single x86-64 CMPQ instruction. Naming convention: "algo" + 2 bytes protocol tag + 2 bytes version

Variables

var ErrInvalidLogLevel = errors.New("invalid log level")

ErrInvalidLogLevel is returned when an invalid log level is provided.

Functions

func EnableP2PLogging

func EnableP2PLogging(log logging.Logger, l logging.Level) error

EnableP2PLogging enables libp2p logging into the provided logger with the provided level.

func GetPeerTelemetryInfo

func GetPeerTelemetryInfo(peerProtocols []protocol.ID) (telemetryID string, telemetryInstance string)

GetPeerTelemetryInfo returns the telemetry ID of a peer by looking at its protocols

func GetPrivKey

func GetPrivKey(cfg config.Local, dataDir string) (crypto.PrivKey, error)

GetPrivKey manages loading and creation of private keys for network PeerIDs It prioritizes, in this order:

1. user supplied path to privKey
2. default path to privKey,
3. generating a new privKey.

If a new privKey is generated it will be saved to default path if cfg.P2PPersistPeerID.

func MakeHost

func MakeHost(cfg config.Local, datadir string, pstore *pstore.PeerStore) (host.Host, string, error)

MakeHost creates a libp2p host but does not start listening. Use host.Network().Listen() on the returned address to start listening.

func MakeService

func MakeService(ctx context.Context, log logging.Logger, cfg config.Local, h host.Host, listenAddr string, wsStreamHandler StreamHandler, metricsTracer pubsub.RawTracer) (*serviceImpl, error)

MakeService creates a P2P service instance

func MakeTestHTTPClient

func MakeTestHTTPClient(addrInfo *peer.AddrInfo, opts ...httpClientOption) (*http.Client, error)

MakeTestHTTPClient creates a http.Client that uses libp2p transport for a given protocol and peer address. This exported method is only used in tests.

func SetP2PLogLevel

func SetP2PLogLevel(l logging.Level) error

SetP2PLogLevel sets the log level for libp2p logging.

func WithHost

func WithHost(h host.Host) httpClientOption

WithHost sets the libp2p host for the http client.

Types

type CapabilitiesDiscovery

type CapabilitiesDiscovery struct {
	// contains filtered or unexported fields
}

CapabilitiesDiscovery exposes Discovery interfaces and wraps underlying DHT methods to provide capabilities advertisement for the node

func MakeCapabilitiesDiscovery

func MakeCapabilitiesDiscovery(ctx context.Context, cfg config.Local, h host.Host, networkID protocol.NetworkID, log logging.Logger, bootstrapFunc func() []peer.AddrInfo) (*CapabilitiesDiscovery, error)

MakeCapabilitiesDiscovery creates a new CapabilitiesDiscovery object which exposes peer discovery and capabilities advertisement

func (*CapabilitiesDiscovery) AdvertiseCapabilities

func (c *CapabilitiesDiscovery) AdvertiseCapabilities(capabilities ...Capability)

AdvertiseCapabilities periodically runs the Advertiser interface on the DHT If a capability fails to advertise we will retry every 10 seconds until full success This gets rerun every at the minimum ttl or the maxAdvertisementInterval.

func (*CapabilitiesDiscovery) Close

func (c *CapabilitiesDiscovery) Close() error

Close should be called when fully shutting down the node

func (*CapabilitiesDiscovery) Host

func (c *CapabilitiesDiscovery) Host() host.Host

Host exposes the underlying libp2p host.Host object

func (*CapabilitiesDiscovery) PeersForCapability

func (c *CapabilitiesDiscovery) PeersForCapability(capability Capability, n int) ([]peer.AddrInfo, error)

PeersForCapability returns a slice of peer.AddrInfo for a Capability Since CapabilitiesDiscovery uses a backoffcache, it will attempt to hit cache, then disk, then network in order to fetch n peers which are advertising the required capability.

func (*CapabilitiesDiscovery) RoutingTable

func (c *CapabilitiesDiscovery) RoutingTable() Sizer

RoutingTable exposes some knowledge about the DHT routing table

type Capability

type Capability string

Capability represents functions that some nodes may provide and other nodes would want to know about

type HTTPServer

type HTTPServer struct {
	libp2phttp.Host
	// contains filtered or unexported fields
}

HTTPServer is a wrapper around libp2phttp.Host that allows registering http handlers with path parameters.

func MakeHTTPServer

func MakeHTTPServer(streamHost host.Host) *HTTPServer

MakeHTTPServer creates a new HTTPServer

func (*HTTPServer) RegisterHTTPHandler

func (s *HTTPServer) RegisterHTTPHandler(path string, handler http.Handler)

RegisterHTTPHandler registers a http handler with a given path.

func (*HTTPServer) RegisterHTTPHandlerFunc

func (s *HTTPServer) RegisterHTTPHandlerFunc(path string, handler func(http.ResponseWriter, *http.Request))

RegisterHTTPHandlerFunc registers a http handler with a given path.

type PeerID

type PeerID string

PeerID is a string representation of a peer's public key, primarily used to avoid importing libp2p into packages that shouldn't need it

func PeerIDFromPublicKey

func PeerIDFromPublicKey(pubKey crypto.PubKey) (PeerID, error)

PeerIDFromPublicKey returns a PeerID from a public key, thin wrapper over libp2p function doing the same

func (PeerID) String

func (id PeerID) String() string

type PeerIDChallengeSigner

type PeerIDChallengeSigner struct {
	// contains filtered or unexported fields
}

PeerIDChallengeSigner implements the identityChallengeSigner interface in the network package.

func (*PeerIDChallengeSigner) PublicKey

func (p *PeerIDChallengeSigner) PublicKey() algocrypto.PublicKey

PublicKey implements the identityChallengeSigner interface.

func (*PeerIDChallengeSigner) Sign

func (p *PeerIDChallengeSigner) Sign(message algocrypto.Hashable) algocrypto.Signature

Sign implements the identityChallengeSigner interface.

func (*PeerIDChallengeSigner) SignBytes

func (p *PeerIDChallengeSigner) SignBytes(message []byte) algocrypto.Signature

SignBytes implements the identityChallengeSigner interface.

type Service

type Service interface {
	Start() error
	Close() error
	ID() peer.ID // return peer.ID for self
	IDSigner() *PeerIDChallengeSigner
	AddrInfo() peer.AddrInfo // return addrInfo for self

	DialPeersUntilTargetCount(targetConnCount int)
	ClosePeer(peer.ID) error

	Conns() []network.Conn
	ListPeersForTopic(topic string) []peer.ID
	Subscribe(topic string, val pubsub.ValidatorEx) (SubNextCancellable, error)
	Publish(ctx context.Context, topic string, data []byte) error

	// GetHTTPClient returns a rate-limiting libp2p-streaming http client that can be used to make requests to the given peer
	GetHTTPClient(addrInfo *peer.AddrInfo, connTimeStore limitcaller.ConnectionTimeStore, queueingTimeout time.Duration) (*http.Client, error)
}

Service defines the interface used by the network integrating with underlying p2p implementation

type Sizer

type Sizer interface {
	Size() int
}

Sizer exposes the Size method

type StreamHandler

type StreamHandler func(ctx context.Context, pid peer.ID, s network.Stream, incoming bool)

StreamHandler is called when a new bidirectional stream for a given protocol and peer is opened.

type SubNextCancellable

type SubNextCancellable interface {
	Next(ctx context.Context) (*pubsub.Message, error)
	Cancel()
}

SubNextCancellable is an abstraction for pubsub.Subscription