merge latest master

libp2p/builders.nim

@@ -15,10 +15,11 @@ import
   crypto/crypto, transports/[transport, tcptransport],
   muxers/[muxer, mplex/mplex],
   protocols/[identify, secure/secure, secure/noise],
-  connmanager, upgrademngrs/muxedupgrade
+  connmanager, upgrademngrs/muxedupgrade,
+  errors
 
 export
-  switch, peerid, peerinfo, connection, multiaddress, crypto
+  switch, peerid, peerinfo, connection, multiaddress, crypto, errors
 
 type
   SecureProtocol* {.pure.} = enum

libp2p/dial.nim

@@ -7,6 +7,8 @@
 ## This file may not be copied, modified, or distributed except according to
 ## those terms.
 
+{.push raises: [Defect].}
+
 import chronos
 import peerid,
        stream/connection

libp2p/dialer.nim

@@ -19,9 +19,10 @@ import dial,
        multistream,
        connmanager,
        stream/connection,
-       transports/transport
+       transports/transport,
+       errors
 
-export dial
+export dial, errors
 
 logScope:
   topics = "libp2p dialer"
@@ -32,7 +33,7 @@ declareCounter(libp2p_failed_dials, "failed dials")
 declareCounter(libp2p_failed_upgrades_outgoing, "outgoing connections failed upgrades")
 
 type
-  DialFailedError* = object of CatchableError
+  DialFailedError* = object of LPError
 
   Dialer* = ref object of Dial
     peerInfo*: PeerInfo

libp2p/muxers/mplex/coder.nim

@@ -9,8 +9,7 @@
 
 {.push raises: [Defect].}
 
-import chronos
-import nimcrypto/utils, chronicles, stew/byteutils
+import pkg/[chronos, nimcrypto/utils, chronicles, stew/byteutils]
 import ../../stream/connection,
        ../../utility,
        ../../varint,
@@ -40,7 +39,7 @@ type
 # https://github.com/libp2p/specs/tree/master/mplex#writing-to-a-stream
 const MaxMsgSize* = 1 shl 20 # 1mb
 
-proc newInvalidMplexMsgType(): ref InvalidMplexMsgType =
+proc newInvalidMplexMsgType*(): ref InvalidMplexMsgType =
   newException(InvalidMplexMsgType, "invalid message type")
 
 proc readMsg*(conn: Connection): Future[Msg] {.async, gcsafe.} =

libp2p/protocols/pubsub/gossipsub.nim

@@ -417,10 +417,24 @@ method publish*(g: GossipSub,
     topic
 
   trace "Publishing message on topic", data = data.shortLog
+
+  if topic.len <= 0: # data could be 0/empty
     debug "Empty topic, skipping publish"
     return 0
 
+  var peers: HashSet[PubSubPeer]
+
+  if g.parameters.floodPublish:
+    # With flood publishing enabled, the mesh is used when propagating messages from other peers,
+    # but a peer's own messages will always be published to all known peers in the topic.
+    for peer in g.gossipsub.getOrDefault(topic):
       if peer.score >= g.parameters.publishThreshold:
+        trace "publish: including flood/high score peer", peer
+        peers.incl(peer)
+
+  # add always direct peers
+  peers.incl(g.explicit.getOrDefault(topic))
+
   if topic in g.topics: # if we're subscribed use the mesh
     peers.incl(g.mesh.getOrDefault(topic))
   else: # not subscribed, send to fanout peers

libp2p/protocols/secure/secio.nim

@@ -439,6 +439,8 @@ proc newSecio*(rng: ref BrHmacDrbgContext, localPrivateKey: PrivateKey): Secio =
   result = Secio(
     rng: rng,
     localPrivateKey: localPrivateKey,
-    localPublicKey: localPrivateKey.getKey().get(),
+    localPublicKey: localPrivateKey
+      .getKey()
+      .expect("Can't fetch local private key"),
   )
   result.init()

libp2p/switch.nim

@@ -123,14 +123,15 @@ proc dial*(
   dial(s, peerId, addrs, @[proto])
 
 proc mount*[T: LPProtocol](s: Switch, proto: T, matcher: Matcher = nil)
-  {.gcsafe, raises: [Defect, CatchableError].} =
+  {.gcsafe, raises: [Defect, LPError].} =
+
   if isNil(proto.handler):
-    raise newException(CatchableError,
-      "Protocol has to define a handle method or proc")
+    raise newException(LPError,
+      "Protocol has to define a `handle` method or proc")
 
   if proto.codec.len == 0:
-    raise newException(CatchableError,
-      "Protocol has to define a codec string")
+    raise newException(LPError,
+      "Protocol has to define a `codec` string")
 
   s.ms.addHandler(proto.codecs, proto, matcher)
   s.peerInfo.protocols.add(proto.codec)
@@ -249,9 +250,10 @@ proc newSwitch*(peerInfo: PeerInfo,
                 muxers: Table[string, MuxerProvider],
                 secureManagers: openarray[Secure] = [],
                 connManager: ConnManager,
-                ms: MultistreamSelect): Switch =
+                ms: MultistreamSelect): Switch
+                {.raises: [Defect, LPError].} =
   if secureManagers.len == 0:
-    raise (ref CatchableError)(msg: "Provide at least one secure manager")
+    raise newException(LPError, "Provide at least one secure manager")
 
   let switch = Switch(
     peerInfo: peerInfo,