Repository: libp2p/go-libp2p
Pull request #2759 (2024-04-05 03:07:08)

Identify: emit useful events after identification
Identify has an event that it emits, but it's not very useful in that it will only tell you a peer has been identified, but not tell you anything about that peer. This design is intended to have the user then lookup peer information somewhere else after they get the event. However, that approach is brittle because it relies on identify itself probing in and updating all the "somewhere elses". This issue is highlight in #2754 and #2355.

Instead, Identify should only concern itself with emitting an event about what it has learned. Other services can subscribe to those events and do their appropriate logic. This helps decouple these services from each other.

Relates to #2754 and #2355 

This is a partial fix for #2754. The next step after this is merged is to update GossipSub to make use of these events.


Patch:
diff --git a/core/event/identify.go b/core/event/identify.go
index ff149958..888572a2 100644
--- a/core/event/identify.go
+++ b/core/event/identify.go
@@ -1,11 +1,40 @@
 package event
 
-import "github.com/libp2p/go-libp2p/core/peer"
+import (
+	"github.com/libp2p/go-libp2p/core/network"
+	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/libp2p/go-libp2p/core/protocol"
+	"github.com/libp2p/go-libp2p/core/record"
+	"github.com/multiformats/go-multiaddr"
+)
 
 // EvtPeerIdentificationCompleted is emitted when the initial identification round for a peer is completed.
 type EvtPeerIdentificationCompleted struct {
 	// Peer is the ID of the peer whose identification succeeded.
 	Peer peer.ID
+
+	// Conn is the connection we identified.
+	Conn network.Conn
+
+	// ListenAddrs is the list of addresses the peer is listening on.
+	ListenAddrs []multiaddr.Multiaddr
+
+	// Protocols is the list of protocols the peer advertised on this connection.
+	Protocols []protocol.ID
+
+	// SignedPeerRecord is the provided signed peer record of the peer. May be nil.
+	SignedPeerRecord *record.Envelope
+
+	// AgentVersion is like a UserAgent string in browsers, or client version in
+	// bittorrent includes the client name and client.
+	AgentVersion string
+
+	// ProtocolVersion is the protocolVersion field in the identify message
+	ProtocolVersion string
+
+	// ObservedAddr is the our side's connection address as observed by the
+	// peer. This is not verified, the peer could return anything here.
+	ObservedAddr multiaddr.Multiaddr
 }
 
 // EvtPeerIdentificationFailed is emitted when the initial identification round for a peer failed.
diff --git a/p2p/protocol/identify/id.go b/p2p/protocol/identify/id.go
index 6c07dbdc..4bb4a592 100644
--- a/p2p/protocol/identify/id.go
+++ b/p2p/protocol/identify/id.go
@@ -400,8 +400,6 @@ func (ids *idService) IdentifyWait(c network.Conn) <-chan struct{} {
 			ids.emitters.evtPeerIdentificationFailed.Emit(event.EvtPeerIdentificationFailed{Peer: c.RemotePeer(), Reason: err})
 			return
 		}
-
-		ids.emitters.evtPeerIdentificationCompleted.Emit(event.EvtPeerIdentificationCompleted{Peer: c.RemotePeer()})
 	}()
 
 	return e.IdentifyWaitChan
@@ -711,8 +709,16 @@ func (ids *idService) consumeMessage(mes *pb.Identify, c network.Conn, isPush bo
 		})
 	}
 
-	// mes.ObservedAddr
-	ids.consumeObservedAddress(mes.GetObservedAddr(), c)
+	obsAddr, err := ma.NewMultiaddrBytes(mes.GetObservedAddr())
+	if err != nil {
+		log.Debugf("error parsing received observed addr for %s: %s", c, err)
+		obsAddr = nil
+	}
+
+	if obsAddr != nil {
+		// TODO refactor this to use the emitted events instead of having this func call explicitly.
+		ids.observedAddrs.Record(c, obsAddr)
+	}
 
 	// mes.ListenAddrs
 	laddrs := mes.GetListenAddrs()
@@ -763,6 +769,7 @@ func (ids *idService) consumeMessage(mes *pb.Identify, c network.Conn, isPush bo
 		signedAddrs, err := ids.consumeSignedPeerRecord(c.RemotePeer(), signedPeerRecord)
 		if err != nil {
 			log.Debugf("failed to consume signed peer record: %s", err)
+			signedPeerRecord = nil
 		} else {
 			addrs = signedAddrs
 		}
@@ -786,6 +793,18 @@ func (ids *idService) consumeMessage(mes *pb.Identify, c network.Conn, isPush bo
 
 	// get the key from the other side. we may not have it (no-auth transport)
 	ids.consumeReceivedPubKey(c, mes.PublicKey)
+
+	ids.emitters.evtPeerIdentificationCompleted.Emit(event.EvtPeerIdentificationCompleted{
+		Peer:             c.RemotePeer(),
+		Conn:             c,
+		ListenAddrs:      lmaddrs,
+		Protocols:        mesProtocols,
+		SignedPeerRecord: signedPeerRecord,
+		ObservedAddr:     obsAddr,
+		ProtocolVersion:  pv,
+		AgentVersion:     av,
+	})
+
 }
 
 func (ids *idService) consumeSignedPeerRecord(p peer.ID, signedPeerRecord *record.Envelope) ([]ma.Multiaddr, error) {
@@ -919,20 +938,6 @@ func HasConsistentTransport(a ma.Multiaddr, green []ma.Multiaddr) bool {
 	return false
 }
 
-func (ids *idService) consumeObservedAddress(observed []byte, c network.Conn) {
-	if observed == nil {
-		return
-	}
-
-	maddr, err := ma.NewMultiaddrBytes(observed)
-	if err != nil {
-		log.Debugf("error parsing received observed addr for %s: %s", c, err)
-		return
-	}
-
-	ids.observedAddrs.Record(c, maddr)
-}
-
 // addConnWithLock assuems caller holds the connsMu lock
 func (ids *idService) addConnWithLock(c network.Conn) {
 	_, found := ids.conns[c]


Test patch:
diff --git a/p2p/protocol/identify/id_test.go b/p2p/protocol/identify/id_test.go
index 0f920356..2352abef 100644
--- a/p2p/protocol/identify/id_test.go
+++ b/p2p/protocol/identify/id_test.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"slices"
 	"sync"
 	"testing"
 	"time"
@@ -200,12 +201,47 @@ func TestIDService(t *testing.T) {
 
 	// test that we received the "identify completed" event.
 	select {
-	case <-sub.Out():
+	case evtAny := <-sub.Out():
+		assertCorrectEvtPeerIdentificationCompleted(t, evtAny, h2)
 	case <-time.After(3 * time.Second):
 		t.Fatalf("expected EvtPeerIdentificationCompleted event within 10 seconds; none received")
 	}
 }
 
+func assertCorrectEvtPeerIdentificationCompleted(t *testing.T, evtAny interface{}, other host.Host) {
+	t.Helper()
+	evt := evtAny.(event.EvtPeerIdentificationCompleted)
+	require.NotNil(t, evt.Conn)
+	require.Equal(t, other.ID(), evt.Peer)
+
+	require.Equal(t, len(other.Addrs()), len(evt.ListenAddrs))
+	if len(other.Addrs()) == len(evt.ListenAddrs) {
+		otherAddrsStrings := make([]string, len(other.Addrs()))
+		evtAddrStrings := make([]string, len(evt.ListenAddrs))
+		for i, a := range other.Addrs() {
+			otherAddrsStrings[i] = a.String()
+			evtAddrStrings[i] = evt.ListenAddrs[i].String()
+		}
+		slices.Sort(otherAddrsStrings)
+		slices.Sort(evtAddrStrings)
+		require.Equal(t, otherAddrsStrings, evtAddrStrings)
+	}
+
+	otherProtos := other.Mux().Protocols()
+	slices.Sort(otherProtos)
+	evtProtos := evt.Protocols
+	slices.Sort(evtProtos)
+	require.Equal(t, otherProtos, evtProtos)
+	idFromSignedRecord, err := peer.IDFromPublicKey(evt.SignedPeerRecord.PublicKey)
+	require.NoError(t, err)
+	require.Equal(t, other.ID(), idFromSignedRecord)
+	require.Equal(t, peer.PeerRecordEnvelopePayloadType, evt.SignedPeerRecord.PayloadType)
+	var peerRecord peer.PeerRecord
+	evt.SignedPeerRecord.TypedRecord(&peerRecord)
+	require.Equal(t, other.ID(), peerRecord.PeerID)
+	require.Equal(t, other.Addrs(), peerRecord.Addrs)
+}
+
 func TestProtoMatching(t *testing.T) {
 	tcp1, _ := ma.NewMultiaddr("/ip4/1.2.3.4/tcp/1234")
 	tcp2, _ := ma.NewMultiaddr("/ip4/1.2.3.4/tcp/2345")
@@ -665,7 +701,8 @@ func TestLargeIdentifyMessage(t *testing.T) {
 
 	// test that we received the "identify completed" event.
 	select {
-	case <-sub.Out():
+	case evtAny := <-sub.Out():
+		assertCorrectEvtPeerIdentificationCompleted(t, evtAny, h2)
 	case <-time.After(3 * time.Second):
 		t.Fatalf("expected EvtPeerIdentificationCompleted event within 3 seconds; none received")
 	}