mesh

README

Mesh

Mesh router transport connects nodes into a full mesh. This is achieved by sharing peers and/or address book with other peers when then successfully handshake.

Protocol

1. Connection completed
2. Periodic ping-pong is sent and connection measured 2.1. No restrictions on the rate and size of ping-pong is emposed
3. Messages can be exchanged 3.1. Only messages that match interest/intent can be shared 3.1. Intent / Interest / Peers / Addrbook messages can be shared

Connection

1. Handshake initiated
2. Connected Peers shared
3. Interests shared
4. Intents shared

The goal of this handshake is to learn about each other FQN. For example a node's configured FQN can be tcp://0.0.0.0:1234/some.path, however, the interface on which it is actually reachable can be tcp://127.0.0.1:1234/some.path or maybe some public IP that is behind NAT. Then both the IP and the address will be different.

In order for the node's address book to be useful, we need to be able to determine what is the ip:port that the node connecting to us is reachable by. Handshake shares this information how it sees the peer with the peer and the peer can adjust its actual address.

The handshake is as follows:

1. Peer sends Handshake message with Initial state 1.1. Peer is initially in INIT state 1.2. Node is in WAITING state 1.3. Me field is peer's FQN (e.g. tcp://0.0.0.0:1234/some.path) 1.4. You field is the node's FQN as known by the peer (e.g. tcp://some.hostname.com:1234/remote.path) 1.5. Intents and Interests will contain routes the peer is interested in 1.6. Peer transitions to WAITING state
2. Node receives handshake, updates peer information, validates 2.1. sends Handshake message with Initial state to the peer 2.2. Node sets Me to its own FQN 2.3. You will contain the peer's FQN as seen by the node 2.4. Intents and Interests will contain routes the node is interested in 2.4. Node transitions to DONE state
3. Peer receives handshake, validates, updates peer information 3.1. sends Handshake message with Final state confirming Me & You fields 3.2. Intents and Interests will contain node's routes the peer successfully registered 3.3. Peer transitions to DONE state
4. Node sends Handshake message with Final state 4.1. Me and You must match advertised respective FQNs 4.2. Intents and Interests will contain peer's routes the node successfully registered 4.3. Node sends Peers message with peers it is connected to
5. Peer receives Peers and adds them to the addrbook 5.1. It will try to connect to these peers if it has no active connection to them yet

Intent

Interest

TODO

• pinging
  • how to get that info?
  • Why though?

FIXME

• transform peer's ip:port so that it can be reachable if it is configured to listen on 0.0.0.0
• do not reset connection when another connection comes in and is rejected
  • reject that new connection sooner
• share new peer with the network when it connects
  • can be periodic

Documentation

Index

• Constants
• type Addrbook
  • func NewAddrbook(self dndm.Peer, peers []*p2ptypes.AddrbookEntry, minConnected int) *Addrbook
  • func (b *Addrbook) ActivePeers(remote dndm.Peer) []string
  • func (b *Addrbook) AddConn(p dndm.Peer, outbound bool, c network.Conn) error
  • func (b *Addrbook) AddPeers(outbound bool, peers ...dndm.Peer) error
  • func (b *Addrbook) Addrbook() []*p2ptypes.AddrbookEntry
  • func (b *Addrbook) DelConn(p dndm.Peer, c network.Conn)
  • func (b *Addrbook) Dials() chan *AddrbookEntry
  • func (b *Addrbook) Entry(p dndm.Peer) (*AddrbookEntry, bool)
  • func (b *Addrbook) Init(ctx context.Context, log *slog.Logger) error
  • func (b *Addrbook) NumConnectedPeers() int
  • func (b *Addrbook) Self() dndm.Peer
  • func (b *Addrbook) SetSharedPeers(remote dndm.Peer, activePeers []string)
• type AddrbookEntry
  • func NewAddrbookEntry(log *slog.Logger, p *p2ptypes.AddrbookEntry) *AddrbookEntry
  • func (a *AddrbookEntry) Dial(ctx context.Context, log *slog.Logger, dialer network.Dialer, ...) (io.ReadWriter, error)
  • func (a *AddrbookEntry) IsActive() bool
  • func (a *AddrbookEntry) SetConn(c network.Conn) error
  • func (a *AddrbookEntry) SetSharedPeers(peers []string)
  • func (a *AddrbookEntry) SharedPeers() map[string]struct{}
  • func (a *AddrbookEntry) WasPeerShared(peer string) bool
• type Container
• type Endpoint
  • func New(localPeer dndm.Peer, size, numDialers int, timeout, pingDuration time.Duration, ...) (*Endpoint, error)
  • func (t *Endpoint) Addrbook() []*p2ptypes.AddrbookEntry
  • func (t *Endpoint) Close() error
  • func (t *Endpoint) Init(ctx context.Context, logger *slog.Logger, addIntent dndm.IntentCallback, ...) error
  • func (t *Endpoint) Local() dndm.Peer
  • func (t *Endpoint) Publish(route dndm.Route, opt ...dndm.PubOpt) (dndm.Intent, error)
  • func (t *Endpoint) Remote() dndm.Peer
  • func (t *Endpoint) Subscribe(route dndm.Route, opt ...dndm.SubOpt) (dndm.Interest, error)
• type HandshakeState
• type Handshaker
  • func NewHandshaker(addrbook *Addrbook, remotePeer dndm.Peer, size int, ...) *Handshaker
  • func (h *Handshaker) Close() error
  • func (h *Handshaker) Init(ctx context.Context, logger *slog.Logger, addIntent dndm.IntentCallback, ...) error
  • func (h *Handshaker) Name() string
  • func (h *Handshaker) OnClose(f func()) dndm.Endpoint
  • func (h *Handshaker) Publish(route dndm.Route, opt ...dndm.PubOpt) (dndm.Intent, error)
  • func (h *Handshaker) SetName(name string)
  • func (h *Handshaker) Subscribe(route dndm.Route, opt ...dndm.SubOpt) (dndm.Interest, error)

Constants

const NumDialers = 10

Types

type Addrbook

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

func NewAddrbook

func NewAddrbook(self dndm.Peer, peers []*p2ptypes.AddrbookEntry, minConnected int) *Addrbook

func (*Addrbook) ActivePeers

func (b *Addrbook) ActivePeers(remote dndm.Peer) []string

ActivePeers returns a list of active peers to be shared to the remote peer. The remote peer itself will be excluded as well as any peers that were already shared with the remote.

func (*Addrbook) AddConn

func (b *Addrbook) AddConn(p dndm.Peer, outbound bool, c network.Conn) error

func (*Addrbook) AddPeers

func (b *Addrbook) AddPeers(outbound bool, peers ...dndm.Peer) error

func (*Addrbook) Addrbook

func (b *Addrbook) Addrbook() []*p2ptypes.AddrbookEntry

func (*Addrbook) DelConn

func (b *Addrbook) DelConn(p dndm.Peer, c network.Conn)

func (*Addrbook) Dials

func (b *Addrbook) Dials() chan *AddrbookEntry

func (*Addrbook) Entry

func (b *Addrbook) Entry(p dndm.Peer) (*AddrbookEntry, bool)

func (*Addrbook) Init

func (b *Addrbook) Init(ctx context.Context, log *slog.Logger) error

func (*Addrbook) NumConnectedPeers

func (b *Addrbook) NumConnectedPeers() int

func (*Addrbook) Self

func (b *Addrbook) Self() dndm.Peer

func (*Addrbook) SetSharedPeers

func (b *Addrbook) SetSharedPeers(remote dndm.Peer, activePeers []string)

SetSharedPeers sets the list of peers that was shared with the target peer. The peers returned by ActivePeers will be excluded from the returned list

type AddrbookEntry

type AddrbookEntry struct {
	sync.Mutex

	Peer              dndm.Peer
	Persistent        bool
	MaxAttempts       int
	DefaultBackoff    time.Duration
	MaxBackoff        time.Duration
	BackoffMultiplier float64

	Attempts    int
	Failed      int
	LastSuccess time.Time
	Backoff     time.Duration
	// contains filtered or unexported fields
}

func NewAddrbookEntry

func NewAddrbookEntry(log *slog.Logger, p *p2ptypes.AddrbookEntry) *AddrbookEntry

func (*AddrbookEntry) Dial

func (a *AddrbookEntry) Dial(ctx context.Context, log *slog.Logger, dialer network.Dialer, q chan<- *AddrbookEntry) (io.ReadWriter, error)

func (*AddrbookEntry) IsActive

func (a *AddrbookEntry) IsActive() bool

func (*AddrbookEntry) SetConn

func (a *AddrbookEntry) SetConn(c network.Conn) error

func (*AddrbookEntry) SetSharedPeers

func (a *AddrbookEntry) SetSharedPeers(peers []string)

func (*AddrbookEntry) SharedPeers

func (a *AddrbookEntry) SharedPeers() map[string]struct{}

func (*AddrbookEntry) WasPeerShared

func (a *AddrbookEntry) WasPeerShared(peer string) bool

type Container

type Container interface {
	Add(dndm.Endpoint) error
	Remove(dndm.Endpoint) error
}

type Endpoint

type Endpoint struct {
	*dndm.Container
	// contains filtered or unexported fields
}

func New

func New(localPeer dndm.Peer, size, numDialers int, timeout, pingDuration time.Duration, node network.Dialer, peers []*p2ptypes.AddrbookEntry) (*Endpoint, error)

func (*Endpoint) Addrbook

func (t *Endpoint) Addrbook() []*p2ptypes.AddrbookEntry

func (*Endpoint) Close

func (t *Endpoint) Close() error

func (*Endpoint) Init

func (t *Endpoint) Init(ctx context.Context, logger *slog.Logger, addIntent dndm.IntentCallback, addInterest dndm.InterestCallback) error

func (*Endpoint) Local

func (t *Endpoint) Local() dndm.Peer

func (*Endpoint) Publish

func (t *Endpoint) Publish(route dndm.Route, opt ...dndm.PubOpt) (dndm.Intent, error)

func (*Endpoint) Remote

func (t *Endpoint) Remote() dndm.Peer

func (*Endpoint) Subscribe

func (t *Endpoint) Subscribe(route dndm.Route, opt ...dndm.SubOpt) (dndm.Interest, error)

type HandshakeState

type HandshakeState int

const (
	HS_INIT HandshakeState = iota
	HS_WAIT
	HS_PEERS
	HS_DONE
)

type Handshaker

type Handshaker struct {
	dndm.BaseEndpoint
	// contains filtered or unexported fields
}

func NewHandshaker

func NewHandshaker(addrbook *Addrbook, remotePeer dndm.Peer, size int, timeout, pingDuration time.Duration, rw io.ReadWriter, state HandshakeState) *Handshaker

NewHandshake creates a Endpoint wrapper over Remote endpoint that acts as a handshake middleware.

initial state transitions:

Connect to remote:
1. HS_INIT
2. HS_WAIT
3. HS_DONE

Remote connects to us:
1. HS_WAIT
2. HS_DONE

func (*Handshaker) Close

func (h *Handshaker) Close() error

func (*Handshaker) Init

func (h *Handshaker) Init(ctx context.Context, logger *slog.Logger, addIntent dndm.IntentCallback, addInterest dndm.InterestCallback) error

Init is used by the Router to initialize this transport.

func (*Handshaker) Name

func (h *Handshaker) Name() string

func (*Handshaker) OnClose

func (h *Handshaker) OnClose(f func()) dndm.Endpoint

func (*Handshaker) Publish

func (h *Handshaker) Publish(route dndm.Route, opt ...dndm.PubOpt) (dndm.Intent, error)

Publish will advertise an intent to publish named and typed data.

func (*Handshaker) SetName

func (h *Handshaker) SetName(name string)

func (*Handshaker) Subscribe

func (h *Handshaker) Subscribe(route dndm.Route, opt ...dndm.SubOpt) (dndm.Interest, error)

Subscribe will advertise an interest in named and typed data.