Reliable Notifiee events

[mrd0ll4r]

Hey there!

I'm using go-libp2p-core v0.9.0 (as an IPFS plugin). I am subscribed to network notifications via the network.Notifiee mechanism:

ipfsInstance.PeerHost.Network().Notify(wiretap)

I don't use the *Stream or Listen* methods, and this is what I currently use to track connected peers on my end:

// WireTap implements the `WireTap` interface to do stuff.
type WireTap struct {
	api         *core.IpfsNode
	connManager connectionManager
	/* ... */
}

type connectionManager struct {
	lock           sync.Mutex
	allConnections map[string]struct{}
	peers          map[peer.ID]peerConnection
}

type peerConnection struct {
	connections   map[string]network.Conn
}

// Connected is called when a connection is opened.
func (wt *WireTap) Connected(_ network.Network, conn network.Conn) {
	remotePeer := conn.RemotePeer()
	log.Debugf("Connection event for PID: %s, Addr: %s\n", remotePeer, conn.RemoteMultiaddr())

	wt.connManager.lock.Lock()
	defer wt.connManager.lock.Unlock()
	if _, ok := wt.connManager.allConnections[conn.ID()]; ok {
		log.Warnf("Duplicate connection ID %s (Duplicate Connection Event)", conn.ID())
		// What now?
	}
	wt.connManager.allConnections[conn.ID()] = struct{}{}

	peerConn, ok := wt.connManager.peers[remotePeer]
	if !ok {
		peerConn = peerConnection{
			connections:   map[string]network.Conn{conn.ID(): conn},
		}
	} else {
		if _, ok := peerConn.connections[conn.ID()]; ok {
			log.Warnf("Duplicate connection ID %s to peer %s", conn.ID(), remotePeer)
			// What now?
		}
		peerConn.connections[conn.ID()] = conn
	}
	wt.connManager.peers[remotePeer] = peerConn
}

// Disconnected is called when a connection is closed.
func (wt *WireTap) Disconnected(_ network.Network, conn network.Conn) {
	remotePeer := conn.RemotePeer()
	log.Debugf("Disconnection event for PID: %s, Addr: %s\n", remotePeer, conn.RemoteMultiaddr())

	wt.connManager.lock.Lock()
	defer wt.connManager.lock.Unlock()
	if _, ok := wt.connManager.allConnections[conn.ID()]; !ok {
		log.Warnf("Missing connection ID %s (Duplicate Disconnection Event or Missing Connection Event)", conn.ID())
		// What now?
	}
	delete(wt.connManager.allConnections, conn.ID())

	peerConn, ok := wt.connManager.peers[remotePeer]
	if !ok {
		log.Warnf("Missing entry for disconnected peer %s", remotePeer)
		// What now?
	} else {
		if _, ok := peerConn.connections[conn.ID()]; !ok {
			log.Warnf("Missing connection entry with ID %s for p %s", conn.ID(), remotePeer)
			// What now?
		}
		delete(peerConn.connections, conn.ID())
		
		if len(peerConn.connections) == 0 {
			delete(wt.connManager.peers, remotePeer)
		}
	}
}

Additionally, I have a goroutine in the background to occasionally print some stats about that:

go func() {
	for {
		peers := wt.api.PeerHost.Peerstore().Peers()
		numConns := 0
		for _, p := range peers {
			conns := wt.api.PeerHost.Network().ConnsToPeer(p)
			numConns += len(conns)
		}

		wt.connManager.lock.Lock()
		numOurPeers := len(wt.connManager.peers)
		numOurConns := len(wt.connManager.allConnections)
		wt.connManager.lock.Unlock()

		log.Infof("We have %d peers. IPFS node has %d connections, we have %d",  numOurPeers, numConns, numOurConns)

		time.Sleep(time.Duration(10) * time.Second)
	}
}()

This is not a 100% accurate comparison, but should be a nice guide.

This all works okay-ish. I occasionally see:

Missing connection ID QmNnooDu7b-3 (Duplicate Disconnection Event or Missing Connection Event)
Missing entry for disconnected peer QmNnooDu7b<redacted>

i.e., we either got a duplicate Disconnected call or didn't get a Connected call.

Additionally, I have noticed that our connection tracking lags behind IPFS', both for connection as well as disconnection events.

The documentation doesn't say much about correctness guarantees, consistency, ... -- What are some properties I can expect?

[marten-seemann]

1. This is not a go-libp2p-core issue.
2. v0.9.0 is really old.
3. You might want to consider subscribing to the EvtPeerConnectednessChanged event on the event bus.