Upgrade libp2p and fix dropped message issue in gossip (#312)

* Add buffer size of pubsub queue

* Fix build error

* Add comment about the reason of bufferSize

go.mod

@@ -7,7 +7,6 @@ require (
 	github.com/btcsuite/btcd v0.21.0-beta
 	github.com/go-kit/kit v0.9.0
 	github.com/golang/protobuf v1.5.2
-	github.com/google/gopacket v1.1.18 // indirect
 	github.com/google/uuid v1.1.4
 	github.com/gorilla/websocket v1.4.2
 	github.com/hashicorp/go-hclog v0.16.2
@@ -17,11 +16,11 @@ require (
 	github.com/hashicorp/hcl v1.0.0
 	github.com/hashicorp/vault/api v1.3.0
 	github.com/imdario/mergo v0.3.7
-	github.com/libp2p/go-libp2p v0.12.0
-	github.com/libp2p/go-libp2p-core v0.7.0
+	github.com/libp2p/go-libp2p v0.14.0
+	github.com/libp2p/go-libp2p-core v0.8.5
 	github.com/libp2p/go-libp2p-kbucket v0.4.7
-	github.com/libp2p/go-libp2p-noise v0.1.1
-	github.com/libp2p/go-libp2p-pubsub v0.4.1
+	github.com/libp2p/go-libp2p-noise v0.2.0
+	github.com/libp2p/go-libp2p-pubsub v0.5.0
 	github.com/mitchellh/cli v1.0.0
 	github.com/multiformats/go-multiaddr v0.3.1
 	github.com/prometheus/client_golang v1.4.0

network/gossip.go

@@ -9,6 +9,13 @@ import (
 	"google.golang.org/protobuf/proto"
 )
 
+const (
+	// bufferSize is the size of the queue in go-libp2p-pubsub
+	// we should have enough capacity of the queue
+	// because there is possibility of that node lose gossip messages when the queue is full
+	bufferSize = 512
+)
+
 type Topic struct {
 	logger hclog.Logger
 
@@ -31,7 +38,7 @@ func (t *Topic) Publish(obj proto.Message) error {
 }
 
 func (t *Topic) Subscribe(handler func(obj interface{})) error {
-	sub, err := t.topic.Subscribe()
+	sub, err := t.topic.Subscribe(pubsub.WithBufferSize(bufferSize))
 	if err != nil {
 		return err
 	}
@@ -54,13 +61,14 @@ func (t *Topic) readLoop(sub *pubsub.Subscription, handler func(obj interface{})
 			t.logger.Error("failed to get topic", "err", err)
 			continue
 		}
-
-		obj := t.createObj()
-		if err := proto.Unmarshal(msg.Data, obj); err != nil {
-			t.logger.Error("failed to unmarshal topic", "err", err)
-			continue
-		}
-		handler(obj)
+		go func() {
+			obj := t.createObj()
+			if err := proto.Unmarshal(msg.Data, obj); err != nil {
+				t.logger.Error("failed to unmarshal topic", "err", err)
+				return
+			}
+			handler(obj)
+		}()
 	}
 }