Merge remote-tracking branch 'origin/unstable' into lesswarnings2

.pinned

@@ -1,16 +1,16 @@
-bearssl;https://github.com/status-im/nim-bearssl@#f4c4233de453cb7eac0ce3f3ffad6496295f83ab
+bearssl;https://github.com/status-im/nim-bearssl@#a647994910904b0103a05db3a5ec1ecfc4d91a88
 chronicles;https://github.com/status-im/nim-chronicles@#32ac8679680ea699f7dbc046e8e0131cac97d41a
-chronos;https://github.com/status-im/nim-chronos@#6525f4ce1d1a7eba146e5f1a53f6f105077ae686
+chronos;https://github.com/status-im/nim-chronos@#75d030ff71264513fb9701c75a326cd36fcb4692
 dnsclient;https://github.com/ba0f3/dnsclient.nim@#fcd7443634b950eaea574e5eaa00a628ae029823
-faststreams;https://github.com/status-im/nim-faststreams@#6112432b3a81d9db116cd5d64c39648881cfff29
-httputils;https://github.com/status-im/nim-http-utils@#e88e231dfcef4585fe3b2fbd9b664dbd28a88040
-json_serialization;https://github.com/status-im/nim-json-serialization@#e5b18fb710c3d0167ec79f3b892f5a7a1bc6d1a4
-metrics;https://github.com/status-im/nim-metrics@#0a6477268e850d7bc98347b3875301524871765f
+faststreams;https://github.com/status-im/nim-faststreams@#b42daf41d8eb4fbce40add6836bed838f8d85b6f
+httputils;https://github.com/status-im/nim-http-utils@#a85bd52ae0a956983ca6b3267c72961d2ec0245f
+json_serialization;https://github.com/status-im/nim-json-serialization@#a7d815ed92f200f490c95d3cfd722089cc923ce6
+metrics;https://github.com/status-im/nim-metrics@#21e99a2e9d9f80e68bef65c80ef781613005fccb
 nimcrypto;https://github.com/cheatfate/nimcrypto@#24e006df85927f64916e60511620583b11403178
-secp256k1;https://github.com/status-im/nim-secp256k1@#c7f1a37d9b0f17292649bfed8bf6cef83cf4221f
-serialization;https://github.com/status-im/nim-serialization@#60a5bd8ac0461dfadd3069fd9c01a7734f205995
-stew;https://github.com/status-im/nim-stew@#23da07c9b59c0ba3d4efa7e4e6e2c4121ae5a156
+secp256k1;https://github.com/status-im/nim-secp256k1@#fd173fdff863ce2e211cf64c9a03bc7539fe40b0
+serialization;https://github.com/status-im/nim-serialization@#d77417cba6896c26287a68e6a95762e45a1b87e5
+stew;https://github.com/status-im/nim-stew@#7184d2424dc3945657884646a72715d494917aad
 testutils;https://github.com/status-im/nim-testutils@#dfc4c1b39f9ded9baf6365014de2b4bfb4dafc34
 unittest2;https://github.com/status-im/nim-unittest2@#da8398c45cafd5bd7772da1fc96e3924a18d3823
-websock;https://github.com/status-im/nim-websock@#acbe30e9ca1e51dcbbfe4c552ee6f16c7eede538
+websock;https://github.com/status-im/nim-websock@#691f069b209d372b1240d5ae1f57fb7bbafeaba7
 zlib;https://github.com/status-im/nim-zlib@#6a6670afba6b97b29b920340e2641978c05ab4d8

examples/README.md

@@ -1,6 +1,5 @@
-# nim-libp2p documentation
+# nim-libp2p examples
 
-Welcome to the nim-libp2p documentation!
+In this folder, you'll find the sources of the [nim-libp2p website](https://status-im.github.io/nim-libp2p/docs/)
 
-Here, you'll find [tutorials](tutorial_1_connect.md) to help you get started, as well as 
-the [full reference](https://status-im.github.io/nim-libp2p/master/libp2p.html).
+We recommand to follow the tutorials on the website, but feel free to grok the sources here!

examples/index.md

@@ -0,0 +1,6 @@
+# nim-libp2p documentation
+
+Welcome to the nim-libp2p documentation!
+
+Here, you'll find [tutorials](tutorial_1_connect.md) to help you get started, as well as 
+the [full reference](https://status-im.github.io/nim-libp2p/master/libp2p.html).

examples/tutorial_2_customproto.nim

@@ -32,7 +32,7 @@ proc new(T: typedesc[TestProto]): T =
     # We must close the connections ourselves when we're done with it
     await conn.close()
 
-  return T(codecs: @[TestCodec], handler: handle)
+  return T.new(codecs = @[TestCodec], handler = handle)
 
 ## This is a constructor for our `TestProto`, that will specify our `codecs` and a `handler`, which will be called for each incoming peer asking for this protocol.
 ## In our handle, we simply read a message from the connection and `echo` it.

examples/tutorial_3_protobuf.nim

@@ -107,16 +107,16 @@ type
     metricGetter: MetricCallback
 
 proc new(_: typedesc[MetricProto], cb: MetricCallback): MetricProto =
-  let res = MetricProto(metricGetter: cb)
+  var res: MetricProto
   proc handle(conn: Connection, proto: string) {.async, gcsafe.} =
     let
       metrics = await res.metricGetter()
       asProtobuf = metrics.encode()
     await conn.writeLp(asProtobuf.buffer)
     await conn.close()
 
-  res.codecs = @["/metric-getter/1.0.0"]
-  res.handler = handle
+  res = MetricProto.new(@["/metric-getter/1.0.0"], handle)
+  res.metricGetter = cb
   return res
 
 proc fetch(p: MetricProto, conn: Connection): Future[MetricList] {.async.} =

examples/tutorial_5_discovery.nim

@@ -36,7 +36,7 @@ proc new(T: typedesc[DumbProto], nodeNumber: int): T =
   proc handle(conn: Connection, proto: string) {.async, gcsafe.} =
     echo "Node", nodeNumber, " received: ", string.fromBytes(await conn.readLp(1024))
     await conn.close()
-  return T(codecs: @[DumbCodec], handler: handle)
+  return T.new(codecs = @[DumbCodec], handler = handle)
 
 ## ## Bootnodes
 ## The first time a p2p program is ran, he needs to know how to join

examples/tutorial_6_game.nim

@@ -157,7 +157,7 @@ proc new(T: typedesc[GameProto], g: Game): T =
     # The handler of a protocol must wait for the stream to
     # be finished before returning
     await conn.join()
-  return T(codecs: @["/tron/1.0.0"], handler: handle)
+  return T.new(codecs = @["/tron/1.0.0"], handler = handle)
 
 proc networking(g: Game) {.async.} =
   # Create our switch, similar to the GossipSub example and

libp2p.nimble

@@ -139,9 +139,13 @@ task install_pinned, "Reads the lockfile":
 
   # Remove the automatically installed deps
   # (inefficient you say?)
-  let allowedDirectories = toInstall.mapIt(it[0] & "-" & it[1].split('@')[1])
-  for dependency in listDirs("nimbledeps/pkgs"):
-    if dependency.extractFilename notin allowedDirectories:
+  let nimblePkgs =
+    if system.dirExists("nimbledeps/pkgs"): "nimbledeps/pkgs"
+    else: "nimbledeps/pkgs2"
+  for dependency in listDirs(nimblePkgs):
+    let filename = dependency.extractFilename
+    if toInstall.anyIt(filename.startsWith(it[0]) and
+       filename.endsWith(it[1].split('#')[^1])) == false:
       rmDir(dependency)
 
 task unpin, "Restore global package use":

libp2p/builders.nim

@@ -61,6 +61,7 @@ type
     autonat: bool
     circuitRelay: Relay
     rdv: RendezVous
+    services: seq[Service]
 
 proc new*(T: type[SwitchBuilder]): T {.public.} =
   ## Creates a SwitchBuilder
@@ -199,6 +200,10 @@ proc withRendezVous*(b: SwitchBuilder, rdv: RendezVous = RendezVous.new()): Swit
   b.rdv = rdv
   b
 
+proc withServices*(b: SwitchBuilder, services: seq[Service]): SwitchBuilder =
+  b.services = services
+  b
+
 proc build*(b: SwitchBuilder): Switch
   {.raises: [Defect, LPError], public.} =
 
@@ -254,7 +259,8 @@ proc build*(b: SwitchBuilder): Switch
     connManager = connManager,
     ms = ms,
     nameResolver = b.nameResolver,
-    peerStore = peerStore)
+    peerStore = peerStore,
+    services = b.services)
 
   if b.autonat:
     let autonat = Autonat.new(switch)

libp2p/connmanager.nim

@@ -110,6 +110,13 @@ proc new*(C: type ConnManager,
 proc connCount*(c: ConnManager, peerId: PeerId): int =
   c.conns.getOrDefault(peerId).len
 
+proc connectedPeers*(c: ConnManager, dir: Direction): seq[PeerId] =
+  var peers = newSeq[PeerId]()
+  for peerId, conns in c.conns:
+    if conns.anyIt(it.dir == dir):
+      peers.add(peerId)
+  return peers
+
 proc addConnEventHandler*(c: ConnManager,
                           handler: ConnEventHandler,
                           kind: ConnEventKind) =
@@ -537,3 +544,4 @@ proc close*(c: ConnManager) {.async.} =
       await conn.close()
 
   trace "Closed ConnManager"
+

libp2p/dial.nim

@@ -36,7 +36,8 @@ method connect*(
 
 method connect*(
   self: Dial,
-  addrs: seq[MultiAddress]): Future[PeerId] {.async, base.} =
+  address: MultiAddress,
+  allowUnknownPeerId = false): Future[PeerId] {.async, base.} =
   ## Connects to a peer and retrieve its PeerId
 
   doAssert(false, "Not implemented!")

libp2p/dialer.nim

@@ -219,11 +219,23 @@ method connect*(
 
 method connect*(
   self: Dialer,
-  addrs: seq[MultiAddress],
-  ): Future[PeerId] {.async.} =
+  address: MultiAddress,
+  allowUnknownPeerId = false): Future[PeerId] {.async.} =
   ## Connects to a peer and retrieve its PeerId
 
-  return (await self.internalConnect(Opt.none(PeerId), addrs, false)).peerId
+  let fullAddress = parseFullAddress(address)
+  if fullAddress.isOk:
+    return (await self.internalConnect(
+      Opt.some(fullAddress.get()[0]),
+      @[fullAddress.get()[1]],
+      false)).peerId
+  else:
+    if allowUnknownPeerId == false:
+      raise newException(DialFailedError, "Address without PeerID and unknown peer id disabled!")
+    return (await self.internalConnect(
+      Opt.none(PeerId),
+      @[address],
+      false)).peerId
 
 proc negotiateStream(
   self: Dialer,

libp2p/multistream.nim

@@ -12,7 +12,7 @@ when (NimMajor, NimMinor) < (1, 4):
 else:
   {.push raises: [].}
 
-import std/[strutils, sequtils]
+import std/[strutils, sequtils, tables]
 import chronos, chronicles, stew/byteutils
 import stream/connection,
        protocols/protocol
@@ -21,7 +21,7 @@ logScope:
   topics = "libp2p multistream"
 
 const
-  MsgSize* = 64*1024
+  MsgSize* = 1024
   Codec* = "/multistream/1.0.0"
 
   MSCodec* = "\x13" & Codec & "\n"
@@ -33,17 +33,20 @@ type
 
   MultiStreamError* = object of LPError
 
-  HandlerHolder* = object
+  HandlerHolder* = ref object
     protos*: seq[string]
     protocol*: LPProtocol
     match*: Matcher
+    openedStreams: CountTable[PeerId]
 
   MultistreamSelect* = ref object of RootObj
     handlers*: seq[HandlerHolder]
     codec*: string
 
 proc new*(T: typedesc[MultistreamSelect]): T =
-  T(codec: MSCodec)
+  T(
+    codec: MSCodec,
+  )
 
 template validateSuffix(str: string): untyped =
     if str.endsWith("\n"):
@@ -169,9 +172,22 @@ proc handle*(m: MultistreamSelect, conn: Connection, active: bool = false) {.asy
         for h in m.handlers:
           if (not isNil(h.match) and h.match(ms)) or h.protos.contains(ms):
             trace "found handler", conn, protocol = ms
-            await conn.writeLp(ms & "\n")
-            conn.protocol = ms
-            await h.protocol.handler(conn, ms)
+
+            var protocolHolder = h
+            let maxIncomingStreams = protocolHolder.protocol.maxIncomingStreams
+            if protocolHolder.openedStreams.getOrDefault(conn.peerId) >= maxIncomingStreams:
+              debug "Max streams for protocol reached, blocking new stream",
+                conn, protocol = ms, maxIncomingStreams
+              return
+            protocolHolder.openedStreams.inc(conn.peerId)
+            try:
+              await conn.writeLp(ms & "\n")
+              conn.protocol = ms
+              await protocolHolder.protocol.handler(conn, ms)
+            finally:
+              protocolHolder.openedStreams.inc(conn.peerId, -1)
+              if protocolHolder.openedStreams[conn.peerId] == 0:
+                protocolHolder.openedStreams.del(conn.peerId)
             return
         debug "no handlers", conn, protocol = ms
         await conn.write(Na)

libp2p/muxers/yamux/yamux.nim

@@ -414,7 +414,8 @@ method close*(m: Yamux) {.async.} =
   let channels = toSeq(m.channels.values())
   for channel in channels:
     await channel.reset(true)
-  await m.connection.write(YamuxHeader.goAway(NormalTermination))
+  try: await m.connection.write(YamuxHeader.goAway(NormalTermination))
+  except CatchableError as exc: trace "failed to send goAway", msg=exc.msg
   await m.connection.close()
   trace "Closed yamux"
 

libp2p/protocols/connectivity/autonat.nim

@@ -32,6 +32,7 @@ const
 
 type
   AutonatError* = object of LPError
+  AutonatUnreachableError* = object of LPError
 
   MsgType* = enum
     Dial = 0
@@ -45,21 +46,21 @@ type
     InternalError = 300
 
   AutonatPeerInfo* = object
-    id: Option[PeerId]
-    addrs: seq[MultiAddress]
+    id*: Option[PeerId]
+    addrs*: seq[MultiAddress]
 
   AutonatDial* = object
-    peerInfo: Option[AutonatPeerInfo]
+    peerInfo*: Option[AutonatPeerInfo]
 
   AutonatDialResponse* = object
     status*: ResponseStatus
     text*: Option[string]
     ma*: Option[MultiAddress]
 
-  AutonatMsg = object
-    msgType: MsgType
-    dial: Option[AutonatDial]
-    response: Option[AutonatDialResponse]
+  AutonatMsg* = object
+    msgType*: MsgType
+    dial*: Option[AutonatDial]
+    response*: Option[AutonatDialResponse]
 
 proc encode*(msg: AutonatMsg): ProtoBuffer =
   result = initProtoBuffer()
@@ -119,7 +120,7 @@ proc encode*(r: AutonatDialResponse): ProtoBuffer =
   result.write(3, bufferResponse.buffer)
   result.finish()
 
-proc decode(_: typedesc[AutonatMsg], buf: seq[byte]): Option[AutonatMsg] =
+proc decode*(_: typedesc[AutonatMsg], buf: seq[byte]): Option[AutonatMsg] =
   var
     msgTypeOrd: uint32
     pbDial: ProtoBuffer
@@ -202,31 +203,45 @@ type
   Autonat* = ref object of LPProtocol
     sem: AsyncSemaphore
     switch*: Switch
+    dialTimeout: Duration
+
+method dialMe*(a: Autonat, pid: PeerId, addrs: seq[MultiAddress] = newSeq[MultiAddress]()):
+    Future[MultiAddress] {.base, async.} =
+
+  proc getResponseOrRaise(autonatMsg: Option[AutonatMsg]): AutonatDialResponse {.raises: [UnpackError, AutonatError].} =
+    if autonatMsg.isNone() or
+       autonatMsg.get().msgType != DialResponse or
+       autonatMsg.get().response.isNone() or
+       (autonatMsg.get().response.get().status == Ok and
+        autonatMsg.get().response.get().ma.isNone()):
+      raise newException(AutonatError, "Unexpected response")
+    else:
+      autonatMsg.get().response.get()
+
+  let conn =
+    try:
+      if addrs.len == 0:
+        await a.switch.dial(pid, @[AutonatCodec])
+      else:
+        await a.switch.dial(pid, addrs, AutonatCodec)
+    except CatchableError as err:
+      raise newException(AutonatError, "Unexpected error when dialling", err)
 
-proc dialMe*(a: Autonat, pid: PeerId, ma: MultiAddress|seq[MultiAddress]):
-    Future[MultiAddress] {.async.} =
-  let addrs = when ma is MultiAddress: @[ma] else: ma
-  let conn = await a.switch.dial(pid, addrs, AutonatCodec)
   defer: await conn.close()
   await conn.sendDial(a.switch.peerInfo.peerId, a.switch.peerInfo.addrs)
-  let msgOpt = AutonatMsg.decode(await conn.readLp(1024))
-  if msgOpt.isNone() or
-     msgOpt.get().msgType != DialResponse or
-     msgOpt.get().response.isNone():
-    raise newException(AutonatError, "Unexpected response")
-  let response = msgOpt.get().response.get()
-  if response.status != ResponseStatus.Ok:
-    raise newException(AutonatError, "Bad status " &
-                                      $response.status & " " &
-                                      response.text.get(""))
-  if response.ma.isNone():
-    raise newException(AutonatError, "Missing address")
-  return response.ma.get()
+  let response = getResponseOrRaise(AutonatMsg.decode(await conn.readLp(1024)))
+  return case response.status:
+    of ResponseStatus.Ok:
+      response.ma.get()
+    of ResponseStatus.DialError:
+      raise newException(AutonatUnreachableError, "Peer could not dial us back")
+    else:
+      raise newException(AutonatError, "Bad status " & $response.status & " " & response.text.get(""))
 
 proc tryDial(a: Autonat, conn: Connection, addrs: seq[MultiAddress]) {.async.} =
   try:
     await a.sem.acquire()
-    let ma = await a.switch.dialer.tryDial(conn.peerId, addrs)
+    let ma = await a.switch.dialer.tryDial(conn.peerId, addrs).wait(a.dialTimeout)
     if ma.isSome:
       await conn.sendResponseOk(ma.get())
     else:
@@ -284,8 +299,8 @@ proc handleDial(a: Autonat, conn: Connection, msg: AutonatMsg): Future[void] =
     return conn.sendResponseError(DialRefused, "No dialable address")
   return a.tryDial(conn, toSeq(addrs))
 
-proc new*(T: typedesc[Autonat], switch: Switch, semSize: int = 1): T =
-  let autonat = T(switch: switch, sem: newAsyncSemaphore(semSize))
+proc new*(T: typedesc[Autonat], switch: Switch, semSize: int = 1, dialTimeout = 15.seconds): T =
+  let autonat = T(switch: switch, sem: newAsyncSemaphore(semSize), dialTimeout: dialTimeout)
   autonat.init()
   autonat
 

libp2p/protocols/connectivity/relay/client.nim

@@ -25,6 +25,7 @@ import ./relay,
        ../../../multiaddress,
        ../../../stream/connection
 
+export options
 
 logScope:
   topics = "libp2p relay relay-client"