Repository: libp2p/go-libp2p
Pull request #1037 (2021-01-11 11:42:26)

Close peerstore and document Host Close
For https://github.com/libp2p/go-libp2p/issues/1030.

MichaelMure: LGTM :+1: 
aarshkshah1992: ping @Stebalien 

Patch:
diff --git a/libp2p.go b/libp2p.go
index 97e3eb9c..fdaa269e 100644
--- a/libp2p.go
+++ b/libp2p.go
@@ -52,7 +52,7 @@ func ChainOptions(opts ...Option) Option {
 // - If no peerstore is provided, the host is initialized with an empty
 // peerstore.
 //
-// Canceling the passed context will stop the returned libp2p node.
+// To stop/shutdown the returned libp2p node, the user needs to cancel the passed context and call `Close` on the returned Host.
 func New(ctx context.Context, opts ...Option) (host.Host, error) {
 	return NewWithoutDefaults(ctx, append(opts, FallbackDefaults)...)
 }
diff --git a/p2p/host/basic/basic_host.go b/p2p/host/basic/basic_host.go
index d69751b0..69baafc8 100644
--- a/p2p/host/basic/basic_host.go
+++ b/p2p/host/basic/basic_host.go
@@ -1007,6 +1007,10 @@ func (h *BasicHost) Close() error {
 		_ = h.emitters.evtLocalAddrsUpdated.Close()
 		h.Network().Close()
 
+		if h.Peerstore() != nil {
+			h.Peerstore().Close()
+		}
+
 		h.refCount.Wait()
 	})
 
diff --git a/p2p/protocol/identify/id.go b/p2p/protocol/identify/id.go
index c245d568..1458152b 100644
--- a/p2p/protocol/identify/id.go
+++ b/p2p/protocol/identify/id.go
@@ -46,6 +46,9 @@ const LibP2PVersion = "ipfs/0.1.0"
 // Deprecated: Set this with the UserAgent option.
 var ClientVersion = "github.com/libp2p/go-libp2p"
 
+// StreamReadTimeout is the read timeout on all incoming Identify family streams.
+var StreamReadTimeout = 60 * time.Second
+
 var (
 	legacyIDSize = 2 * 1024 // 2k Bytes
 	signedIDSize = 8 * 1024 // 8K
@@ -369,7 +372,8 @@ func (ids *IDService) identifyConn(c network.Conn, signal chan struct{}) {
 		s.Reset()
 		return
 	}
-	ids.handleIdentifyResponse(s)
+
+	err = ids.handleIdentifyResponse(s)
 }
 
 func (ids *IDService) sendIdentifyResp(s network.Stream) {
@@ -408,7 +412,9 @@ func (ids *IDService) sendIdentifyResp(s network.Stream) {
 	log.Debugf("%s sent message to %s %s", ID, c.RemotePeer(), c.RemoteMultiaddr())
 }
 
-func (ids *IDService) handleIdentifyResponse(s network.Stream) {
+func (ids *IDService) handleIdentifyResponse(s network.Stream) error {
+	_ = s.SetReadDeadline(time.Now().Add(StreamReadTimeout))
+
 	c := s.Conn()
 
 	r := protoio.NewDelimitedReader(s, signedIDSize)
@@ -417,7 +423,7 @@ func (ids *IDService) handleIdentifyResponse(s network.Stream) {
 	if err := readAllIDMessages(r, mes); err != nil {
 		log.Warning("error reading identify message: ", err)
 		s.Reset()
-		return
+		return err
 	}
 
 	defer s.Close()
@@ -425,6 +431,8 @@ func (ids *IDService) handleIdentifyResponse(s network.Stream) {
 	log.Debugf("%s received message from %s %s", s.Protocol(), c.RemotePeer(), c.RemoteMultiaddr())
 
 	ids.consumeMessage(mes, c)
+
+	return nil
 }
 
 func readAllIDMessages(r protoio.Reader, finalMsg proto.Message) error {
diff --git a/p2p/protocol/identify/id_delta.go b/p2p/protocol/identify/id_delta.go
index 089f722c..48bae8bf 100644
--- a/p2p/protocol/identify/id_delta.go
+++ b/p2p/protocol/identify/id_delta.go
@@ -5,6 +5,7 @@ import (
 	"github.com/libp2p/go-libp2p-core/network"
 	"github.com/libp2p/go-libp2p-core/peer"
 	"github.com/libp2p/go-libp2p-core/protocol"
+	"time"
 
 	pb "github.com/libp2p/go-libp2p/p2p/protocol/identify/pb"
 
@@ -15,6 +16,8 @@ const IDDelta = "/p2p/id/delta/1.0.0"
 
 // deltaHandler handles incoming delta updates from peers.
 func (ids *IDService) deltaHandler(s network.Stream) {
+	_ = s.SetReadDeadline(time.Now().Add(StreamReadTimeout))
+
 	c := s.Conn()
 
 	r := protoio.NewDelimitedReader(s, 2048)


Test patch:
diff --git a/p2p/protocol/identify/id_test.go b/p2p/protocol/identify/id_test.go
index 2c838bc2..326ccea7 100644
--- a/p2p/protocol/identify/id_test.go
+++ b/p2p/protocol/identify/id_test.go
@@ -978,3 +978,83 @@ func TestLargePushMessage(t *testing.T) {
 	}, 5*time.Second, 500*time.Millisecond)
 	testHasCertifiedAddrs(t, h2, h1p, h1.Addrs())
 }
+
+func TestIdentifyResponseReadTimeout(t *testing.T) {
+	ctx := context.Background()
+	timeout := identify.StreamReadTimeout
+	identify.StreamReadTimeout = 100 * time.Millisecond
+	defer func() {
+		identify.StreamReadTimeout = timeout
+	}()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	h1 := blhost.NewBlankHost(swarmt.GenSwarm(t, ctx))
+	h2 := blhost.NewBlankHost(swarmt.GenSwarm(t, ctx))
+	defer h1.Close()
+	defer h2.Close()
+
+	h2p := h2.ID()
+	ids1 := identify.NewIDService(h1)
+	ids2 := identify.NewIDService(h2)
+	defer ids1.Close()
+	defer ids2.Close()
+	// remote stream handler will just hang and not send back an identify response
+	h2.SetStreamHandler(identify.ID, func(s network.Stream) {
+		time.Sleep(100 * time.Second)
+	})
+
+	sub, err := ids1.Host.EventBus().Subscribe(new(event.EvtPeerIdentificationFailed), eventbus.BufSize(16))
+	require.NoError(t, err)
+
+	h2pi := h2.Peerstore().PeerInfo(h2p)
+	require.NoError(t, h1.Connect(ctx, h2pi))
+
+	select {
+	case ev := <-sub.Out():
+		fev := ev.(event.EvtPeerIdentificationFailed)
+		require.EqualError(t, fev.Reason, "i/o deadline reached")
+	case <-time.After(5 * time.Second):
+		t.Fatal("did not receive identify failure event")
+	}
+}
+
+func TestIncomingIDStreamsTimeout(t *testing.T) {
+	ctx := context.Background()
+	timeout := identify.StreamReadTimeout
+	identify.StreamReadTimeout = 100 * time.Millisecond
+	defer func() {
+		identify.StreamReadTimeout = timeout
+	}()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	protocols := []protocol.ID{identify.IDPush, identify.IDDelta}
+
+	for _, p := range protocols {
+		h1 := blhost.NewBlankHost(swarmt.GenSwarm(t, ctx))
+		h2 := blhost.NewBlankHost(swarmt.GenSwarm(t, ctx))
+		defer h1.Close()
+		defer h2.Close()
+
+		ids1 := identify.NewIDService(h1)
+		ids2 := identify.NewIDService(h2)
+		defer ids1.Close()
+		defer ids2.Close()
+
+		h2p := h2.ID()
+		h2pi := h2.Peerstore().PeerInfo(h2p)
+		require.NoError(t, h1.Connect(ctx, h2pi))
+
+		_, err := h1.NewStream(ctx, h2p, p)
+		require.NoError(t, err)
+
+		// remote peer should eventually reset stream
+		require.Eventually(t, func() bool {
+			c := h2.Network().ConnsToPeer(h1.ID())[0]
+			return len(c.GetStreams()) == 0
+		}, 1*time.Second, 200*time.Millisecond)
+	}
+}