celestiaorg / go-libp2p-messenger

Messenger provides a simple arbitrary message sending API to multiple peers for libp2p-based protocols.

Messenger

Messenger provides a simple arbitrary message sending API to multiple peers for libp2p-based protocols. The main purpose is to bootstrap development for new protocols and decrease boilerplate code.

Features

• Simple non-blocking API
• Not a protocol - sends only user's data without message wrapping on the wire
• Not a framework - use it anyhow
• Integrates generics in a clean way
• Almost zero allocations API
• Hides away stream management(re-creation, duplicates, lifecycle)

Background

The libp2p provides all necessary primitives to build custom fully decentralized p2p protocols. Also, it is solely stream-based for reasons, but in practice, the vast majority of libp2p based protocols are message-based, with messages sent over reliable and multiplexed streams.

Mainly, there are two typical patterns for message sending over streams.

Request/Response

The most popular pattern in libp2p protocols is sending requests over a fresh stream and waiting for a response with a subsequent closing. Using libp2p's bare Stream API is trivial for such a case, which also explains why it is the most popular pattern.

'Fire and Forget'

Another less common pattern in libp2p protocols is sending messages without expecting a response. Once instantiated, a protocol establishes a unique and persistent stream with a remote endpoint and only sends messages over it. This pattern optimizes protocols by removing unnecessary stream negotiations, which are primarily expensive for round trips.

Motivations

• Provide reusable utility for common use-cases in libp2p protocol development.

  • Even for simple Request/Response pattern, there is some level of boilerplate that can be avoided (Examples: Bitswap, kadDHT)

    Currently, the Messenger does not provide a Request/Response semantics, and the main focus is 'fire and forget'.

  • 'Fire and Forget' is more complex. It can be implemented in multiple ways, has undesirable edge cases, and requires a more profound knowledge of the libp2p stack. Although the pattern is an interesting engineering challenge to solve, in business and rapidly changing environments, it is easier to reuse existing instruments, like Messenger.

• Lower entrance threshold for new developers exploring decentralization and p2p networking.

• Bootstrap migration of Tendermint over libp2p.

  Tendermint is the first and most popular BFT consensus implementation, and it is decentralized and requires p2p networking. The project's team, through time, developed an in-house solution for it. However, they are now committed to migration over libp2p.

  Tendermint is a towering stack of multiple sub-protocols(or their terms - reactors) which rely on patterns explained above. Therefore, a lesser code boilerplate and simple API would be convenient for such migration.

  Technically, Tendermint p2p stack only supports 'fire and forget' pattern, but a few sub-protocols can benefit from migration to Request/Response pattern.

Documentation

See these docs.

License

Apache-2.0.

go-libp2p-messenger's Issues

Per peer msg handling

Context

Messenger only provides Send and Receive for msg handling and both methods are unified for any peer. This forces protocols to be implemented in one way only, where msgs are sent or rcvd in only one place. However, there are protocols that would benefit from splitting logic per each peer independently.

Design

// The peer structure stores required channels to send/recv msgs
type Peer struct {}
// Send mimics Messenger's Send but without peer id.
func (p *Peer) Send(context.Context, serde.Message) <-chan err
// Receive mimics Messenger's Receive but without peer id.
// Multiple Receive users will compete for msgs, as they are not duplicated.
func (p *Peer) Receive(context.Context) serde.Message
// Events chan will only return event's of the peer.
func (p *Peer) Events() <-chan PeerEvent

// Add a new method to the Messenger to return a Peer.
// Once a Peer instantiated, new msgs incoming from the peer can only be rcvd through Peer.Receive. 
// Messenger.Receive will still be able to receive new msgs from other peers, but not from this one.
func (m *Messenger Peer(peer.ID) *Peer

Extend Messenger API to handle multiple msg types at once

Context

Currently, Messenger allows only one serde.Message type per itself. However, there are protocols that want to have multiple msg types. This may also mean having separate streams and queues for each msg type, so they won't block each other. However, still having only one stream per multiple message types may open room for msg prioritization via a priority queue.

Improve Readme by explaing why Streams are great, but can be simplied even further

Requestor

Context

As explained in README, the most popular libp2p pattern is request/response and the goal here is to provide utility for it.

Design

type Requestor struct { 
// some private fields 
}
// NewRequestor instantiates new Requestor for a given protocol ID with specified request and response
// message types.
NewRequestor(protocol.Id, serde.Message, serde.Message) *Requestor

// Request sends the request as given message, blocks until a message is written, and returns a       
// channel awaiting for a response. 
// NOTE: This can also be done in fully non-blocking fashion if needed.
func (r *Requestor) Request(peer.ID, serde.Message) <-chan serde.Message, error

Promote usage of EventBus instead of Events; remove custom Events API

Background

Now that libp2p/go-libp2p#1230 exists, it does not make sense to define custom events API, and users should rely on EventBus instead.

Actions

• Remove Events method
• Write a test asserting a random msg being successfully sent after EvtPeerConnectednessChanged is recvd.

Option to tag peers automatically with ConnectionManager

In extensive DHT networks with lots of transient connections ConnectionManager can trim the messenger's connection, and to avoid this, the messenger should tag those connections to some weight.

serde: Allow users to set custom max msg size limit

Background

Currently, there is a global 1mb limit for msgs, but some users(@liamsi) would like to have per-protocol customization.

Implementation ideas

• Extend Write with one more max param (unlikely)
• Add additional interface for msgs(preffered):

// BoundedMessage can be implemented by msgs to set a customized size limit.
type BoundedMessage interface {
	Message
	
	MaxSize() int
}

The Write should then check if msg implements the interface and if it is, limit max allowed msg size.

Update README by removing Tendermint to libp2p migration

Zero alloc version of Broadcast that does return peers msg was sent to

Would also need to write zero alloc proving tests

Don't expose network.Notifee methods

And ideally stop relying on Notifee and use event buts instead

Introduce MultiProtocolMessenger aka Router

Which lazily spawns a new instance of messenger for each new protocol message sent to it and forwards further msgs to the right protocol instance.

Cleanup idle streams

In the same way, we lazily create outbound streams, we can clean up those if there was no msg sent from us in the while. "in-while" should be a configurable value.

Introduce msngr.Message

Which embeds serde.Message with two additional methods:
Protocol() protocol.ID
To() peer.ID

This excludes allocations for message wrapper per each msg on a hot code path. And provides flexibility to add more features to messages via extending it with further (optional) methods.

Generics support

Currently Messenger allows to specify msg type for the protocol by reflection, so that it can construct and give the user already deserialized messages. Since go1.18 there is support for generics which can be used for this specific case instead of generics.

Smarter reconnection handling for msg queues

Context

Messenger manages a per-peer queue of messages. Those queues are not cleaned up right now if a peer disconnects, which is technically a bug.

However, the goal is to allow queues to live for some configurable amount of time if peer disconnects, so once it reconnects the same queue is reused and writing of outstanding msgs restarts.

Unfortunately, there is a chance for messages to be lost, as any msg written to a libp2p stream can return a success, while in fact it was not send. This is happening because of the non-blocking API of stream writes in libp2p. I.e. a case possible where a connection can be closed right after a successful write, but the write itself was buffered and wasn't written to the closed connection.

Therefore, there is no precise way for Messenger to know which message has to be resent on reconnect, because a msg could be written to the stream, return success, but in fact lost forever.

api: redundant error channels on Send/Broadcast

Currently, both Send and Broadcast return the error channel to the user for error handling and as a signal that msg was sent or not. However:

• There is no way to guarantee that the msg was sent, as libp2p streams do not have such a guarantee. That is, if a user writes to a libp2p stream and the write does not return any error, it is not a guarantee that msg was sent on the underlying connection and rcvd by another end. Thus, there is no point in passing the error return after writing (unless the libp2p stream does not provide such a guarantee). Although, this does not mean that errors returned from writes have no value. Mainly, errors returned from streams are about stream lifecycle, e.g., closed or reset, which the Messenger handles underneath without the need to notify the user.
• Creating a channel on each message is expensive, and by removing them, Messanger performance could be improved in protocols with a high msg exchange rate.

Integrate with resource mangager

go-libp2p recently introduced a resource manager which allows fine-grained control of resources(per peer, per stream, per protocol, per conn, etc). The Messanger can potentially also be limited by the resource manager to follow global limts.