Add queued send bytes

Co-authored-by: diegomrsantos <diego@status.im>

.github/actions/install_nim/action.yml

@@ -0,0 +1,131 @@
+name: Install Nim
+inputs:
+  os:
+    description: "Operating system to build for"
+    required: true
+  cpu:
+    description: "CPU to build for"
+    default: "amd64"
+  nim_branch:
+    description: "Nim version"
+    default: "version-1-6"
+  shell:
+    description: "Shell to run commands in"
+    default: "bash --noprofile --norc -e -o pipefail"
+
+runs:
+  using: "composite"
+  steps:
+    - name: Install build dependencies (Linux i386)
+      shell: ${{ inputs.shell }}
+      if: inputs.os == 'Linux' && inputs.cpu == 'i386'
+      run: |
+        sudo dpkg --add-architecture i386
+        sudo apt-get update -qq
+        sudo DEBIAN_FRONTEND='noninteractive' apt-get install \
+          --no-install-recommends -yq gcc-multilib g++-multilib \
+          libssl-dev:i386
+        mkdir -p external/bin
+        cat << EOF > external/bin/gcc
+        #!/bin/bash
+        exec $(which gcc) -m32 "\$@"
+        EOF
+        cat << EOF > external/bin/g++
+        #!/bin/bash
+        exec $(which g++) -m32 "\$@"
+        EOF
+        chmod 755 external/bin/gcc external/bin/g++
+        echo '${{ github.workspace }}/external/bin' >> $GITHUB_PATH
+
+    - name: MSYS2 (Windows i386)
+      if: inputs.os == 'Windows' && inputs.cpu == 'i386'
+      uses: msys2/setup-msys2@v2
+      with:
+        path-type: inherit
+        msystem: MINGW32
+        install: >-
+          base-devel
+          git
+          mingw-w64-i686-toolchain
+
+    - name: MSYS2 (Windows amd64)
+      if: inputs.os == 'Windows' && inputs.cpu == 'amd64'
+      uses: msys2/setup-msys2@v2
+      with:
+        path-type: inherit
+        install: >-
+          base-devel
+          git
+          mingw-w64-x86_64-toolchain
+
+    - name: Restore Nim DLLs dependencies (Windows) from cache
+      if: inputs.os == 'Windows'
+      id: windows-dlls-cache
+      uses: actions/cache@v3
+      with:
+        path: external/dlls
+        key: 'dlls'
+
+    - name: Install DLL dependencies (Windows)
+      shell: ${{ inputs.shell }}
+      if: >
+        steps.windows-dlls-cache.outputs.cache-hit != 'true' &&
+        inputs.os == 'Windows'
+      run: |
+        mkdir external
+        curl -L "https://nim-lang.org/download/windeps.zip" -o external/windeps.zip
+        7z x external/windeps.zip -oexternal/dlls
+
+    - name: Path to cached dependencies (Windows)
+      shell: ${{ inputs.shell }}
+      if: >
+        inputs.os == 'Windows'
+      run: |
+        echo '${{ github.workspace }}'"/external/dlls" >> $GITHUB_PATH
+
+    - name: Derive environment variables
+      shell: ${{ inputs.shell }}
+      run: |
+        if [[ '${{ inputs.cpu }}' == 'amd64' ]]; then
+          PLATFORM=x64
+        else
+          PLATFORM=x86
+        fi
+        echo "PLATFORM=$PLATFORM" >> $GITHUB_ENV
+
+        ncpu=
+        MAKE_CMD="make"
+        case '${{ inputs.os }}' in
+        'Linux')
+          ncpu=$(nproc)
+          ;;
+        'macOS')
+          ncpu=$(sysctl -n hw.ncpu)
+          ;;
+        'Windows')
+          ncpu=$NUMBER_OF_PROCESSORS
+          MAKE_CMD="mingw32-make"
+          ;;
+        esac
+        [[ -z "$ncpu" || $ncpu -le 0 ]] && ncpu=1
+        echo "ncpu=$ncpu" >> $GITHUB_ENV
+        echo "MAKE_CMD=${MAKE_CMD}" >> $GITHUB_ENV
+        echo '${{ github.workspace }}/nim/bin' >> $GITHUB_PATH
+
+    - name: Restore Nim from cache
+      id: nim-cache
+      uses: actions/cache@v3
+      with:
+        path: '${{ github.workspace }}/nim'
+        key: ${{ inputs.os }}-${{ inputs.cpu }}-nim-${{ inputs.nim_branch }}-cache-${{ env.cache_nonce }}
+
+    - name: Build Nim and Nimble
+      shell: ${{ inputs.shell }}
+      if: ${{ steps.nim-cache.outputs.cache-hit != 'true' }}
+      run: |
+        # We don't want partial matches of the cache restored
+        rm -rf nim
+        curl -O -L -s -S https://raw.githubusercontent.com/status-im/nimbus-build-system/master/scripts/build_nim.sh
+        env MAKE="${MAKE_CMD} -j${ncpu}" ARCH_OVERRIDE=${PLATFORM} NIM_COMMIT=${{ inputs.nim_branch }} \
+          QUICK_AND_DIRTY_COMPILER=1 QUICK_AND_DIRTY_NIMBLE=1 CC=gcc \
+          bash build_nim.sh nim csources dist/nimble NimBinaries

.github/workflows/ci.yml

@@ -7,6 +7,10 @@ on:
   pull_request:
   workflow_dispatch:
 
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
 jobs:
   build:
     timeout-minutes: 90
@@ -45,111 +49,20 @@ jobs:
 
     name: '${{ matrix.target.os }}-${{ matrix.target.cpu }} (Nim ${{ matrix.branch }})'
     runs-on: ${{ matrix.builder }}
-    continue-on-error: ${{ matrix.branch == 'version-1-6' || matrix.branch == 'devel' }}
+    continue-on-error: ${{ matrix.branch == 'devel' }}
     steps:
       - name: Checkout
         uses: actions/checkout@v2
         with:
           submodules: true
 
-      - name: Install build dependencies (Linux i386)
-        if: runner.os == 'Linux' && matrix.target.cpu == 'i386'
-        run: |
-          sudo dpkg --add-architecture i386
-          sudo apt-get update -qq
-          sudo DEBIAN_FRONTEND='noninteractive' apt-get install \
-            --no-install-recommends -yq gcc-multilib g++-multilib \
-            libssl-dev:i386
-          mkdir -p external/bin
-          cat << EOF > external/bin/gcc
-          #!/bin/bash
-          exec $(which gcc) -m32 "\$@"
-          EOF
-          cat << EOF > external/bin/g++
-          #!/bin/bash
-          exec $(which g++) -m32 "\$@"
-          EOF
-          chmod 755 external/bin/gcc external/bin/g++
-          echo '${{ github.workspace }}/external/bin' >> $GITHUB_PATH
-
-      - name: MSYS2 (Windows i386)
-        if: runner.os == 'Windows' && matrix.target.cpu == 'i386'
-        uses: msys2/setup-msys2@v2
+      - name: Setup Nim
+        uses: "./.github/actions/install_nim"
         with:
-          path-type: inherit
-          msystem: MINGW32
-          install: >-
-            base-devel
-            git
-            mingw-w64-i686-toolchain
-
-      - name: MSYS2 (Windows amd64)
-        if: runner.os == 'Windows' && matrix.target.cpu == 'amd64'
-        uses: msys2/setup-msys2@v2
-        with:
-          path-type: inherit
-          install: >-
-            base-devel
-            git
-            mingw-w64-x86_64-toolchain
-
-      - name: Restore Nim DLLs dependencies (Windows) from cache
-        if: runner.os == 'Windows'
-        id: windows-dlls-cache
-        uses: actions/cache@v2
-        with:
-          path: external/dlls
-          key: 'dlls'
-
-      - name: Install DLL dependencies (Windows)
-        if: >
-          steps.windows-dlls-cache.outputs.cache-hit != 'true' &&
-          runner.os == 'Windows'
-        run: |
-          mkdir external
-          curl -L "https://nim-lang.org/download/windeps.zip" -o external/windeps.zip
-          7z x external/windeps.zip -oexternal/dlls
-
-      - name: Path to cached dependencies (Windows)
-        if: >
-          runner.os == 'Windows'
-        run: |
-          echo '${{ github.workspace }}'"/external/dlls" >> $GITHUB_PATH
-
-      - name: Derive environment variables
-        run: |
-          if [[ '${{ matrix.target.cpu }}' == 'amd64' ]]; then
-            PLATFORM=x64
-          else
-            PLATFORM=x86
-          fi
-          echo "PLATFORM=$PLATFORM" >> $GITHUB_ENV
-
-          ncpu=
-          MAKE_CMD="make"
-          case '${{ runner.os }}' in
-          'Linux')
-            ncpu=$(nproc)
-            ;;
-          'macOS')
-            ncpu=$(sysctl -n hw.ncpu)
-            ;;
-          'Windows')
-            ncpu=$NUMBER_OF_PROCESSORS
-            MAKE_CMD="mingw32-make"
-            ;;
-          esac
-          [[ -z "$ncpu" || $ncpu -le 0 ]] && ncpu=1
-          echo "ncpu=$ncpu" >> $GITHUB_ENV
-          echo "MAKE_CMD=${MAKE_CMD}" >> $GITHUB_ENV
-
-      - name: Build Nim and Nimble
-        run: |
-          curl -O -L -s -S https://raw.githubusercontent.com/status-im/nimbus-build-system/master/scripts/build_nim.sh
-          env MAKE="${MAKE_CMD} -j${ncpu}" ARCH_OVERRIDE=${PLATFORM} NIM_COMMIT=${{ matrix.branch }} \
-            QUICK_AND_DIRTY_COMPILER=1 QUICK_AND_DIRTY_NIMBLE=1 CC=gcc \
-            bash build_nim.sh nim csources dist/nimble NimBinaries
-          echo '${{ github.workspace }}/nim/bin' >> $GITHUB_PATH
+          os: ${{ matrix.target.os }}
+          cpu: ${{ matrix.target.cpu }}
+          shell: ${{ matrix.shell }}
+          nim_branch: ${{ matrix.branch }}
 
       - name: Setup Go
         uses: actions/setup-go@v2
@@ -160,9 +73,20 @@ jobs:
         run: |
           V=1 bash scripts/build_p2pd.sh p2pdCache 124530a3
 
+      - name: Restore deps from cache
+        id: deps-cache
+        uses: actions/cache@v3
+        with:
+          path: nimbledeps
+          key: nimbledeps-${{ hashFiles('.pinned') }}
+
+      - name: Install deps
+        if: ${{ steps.deps-cache.outputs.cache-hit != 'true' }}
+        run: |
+          nimble install_pinned
+
       - name: Run tests
         run: |
           nim --version
           nimble --version
-          nimble install_pinned
           nimble test

.github/workflows/codecov.yml

@@ -9,6 +9,10 @@ on:
   pull_request:
   workflow_dispatch:
 
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
 jobs:
   Coverage:
     runs-on: ubuntu-20.04
@@ -18,15 +22,31 @@ jobs:
       - uses: actions/checkout@v2
         with:
           fetch-depth: 0
+
+      - name: Setup Nim
+        uses: "./.github/actions/install_nim"
+        with:
+          os: linux
+          cpu: amd64
+          shell: bash
+
+      - name: Restore deps from cache
+        id: deps-cache
+        uses: actions/cache@v3
+        with:
+          path: nimbledeps
+          key: nimbledeps-${{ hashFiles('.pinned') }}
+
+      - name: Install deps
+        if: ${{ steps.deps-cache.outputs.cache-hit != 'true' }}
+        run: |
+          nimble install_pinned
+
       - name: Run
         run: |
           sudo apt-get update
           sudo apt-get install -y lcov build-essential git curl
           mkdir coverage
-          curl -O -L -s -S https://raw.githubusercontent.com/status-im/nimbus-build-system/master/scripts/build_nim.sh
-          env MAKE="make -j${NPROC}" bash build_nim.sh Nim csources dist/nimble NimBinaries
-          export PATH="$PATH:$PWD/Nim/bin"
-          nimble install_pinned
           export NIMFLAGS="--lineDir:on --passC:-fprofile-arcs --passC:-ftest-coverage --passL:-fprofile-arcs --passL:-ftest-coverage"
           nimble testnative
           nimble testpubsub
@@ -38,7 +58,8 @@ jobs:
           lcov --extract coverage/coverage.info  `pwd`/libp2p/{*,**/*}.nim --output-file coverage/coverage.f.info
           genhtml coverage/coverage.f.info --output-directory coverage/output
           bash <(curl -s https://codecov.io/bash) -f coverage/coverage.f.info || echo "Codecov did not collect coverage reports"
-      - uses: actions/upload-artifact@master
-        with:
-          name: coverage
-          path: coverage
+
+          #- uses: actions/upload-artifact@master
+          #  with:
+          #    name: coverage
+          #    path: coverage

.github/workflows/multi_nim.yml

@@ -5,7 +5,13 @@ on:
   workflow_dispatch:
 
 jobs:
+  delete-cache:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: snnaplab/delete-branch-cache-action@v1
+
   build:
+    needs: delete-cache
     timeout-minutes: 120
     strategy:
       fail-fast: false
@@ -46,107 +52,14 @@ jobs:
     steps:
       - name: Checkout
         uses: actions/checkout@v2
+
+      - name: Setup Nim
+        uses: "./.github/actions/install_nim"
         with:
-          submodules: true
-
-      - name: Install build dependencies (Linux i386)
-        if: runner.os == 'Linux' && matrix.target.cpu == 'i386'
-        run: |
-          sudo dpkg --add-architecture i386
-          sudo apt-get update -qq
-          sudo DEBIAN_FRONTEND='noninteractive' apt-get install \
-            --no-install-recommends -yq gcc-multilib g++-multilib \
-            libssl-dev:i386
-          mkdir -p external/bin
-          cat << EOF > external/bin/gcc
-          #!/bin/bash
-          exec $(which gcc) -m32 "\$@"
-          EOF
-          cat << EOF > external/bin/g++
-          #!/bin/bash
-          exec $(which g++) -m32 "\$@"
-          EOF
-          chmod 755 external/bin/gcc external/bin/g++
-          echo '${{ github.workspace }}/external/bin' >> $GITHUB_PATH
-
-      - name: MSYS2 (Windows i386)
-        if: runner.os == 'Windows' && matrix.target.cpu == 'i386'
-        uses: msys2/setup-msys2@v2
-        with:
-          path-type: inherit
-          msystem: MINGW32
-          install: >-
-            base-devel
-            git
-            mingw-w64-i686-toolchain
-
-      - name: MSYS2 (Windows amd64)
-        if: runner.os == 'Windows' && matrix.target.cpu == 'amd64'
-        uses: msys2/setup-msys2@v2
-        with:
-          path-type: inherit
-          install: >-
-            base-devel
-            git
-            mingw-w64-x86_64-toolchain
-
-      - name: Restore Nim DLLs dependencies (Windows) from cache
-        if: runner.os == 'Windows'
-        id: windows-dlls-cache
-        uses: actions/cache@v2
-        with:
-          path: external/dlls
-          key: 'dlls'
-
-      - name: Install DLL dependencies (Windows)
-        if: >
-          steps.windows-dlls-cache.outputs.cache-hit != 'true' &&
-          runner.os == 'Windows'
-        run: |
-          mkdir external
-          curl -L "https://nim-lang.org/download/windeps.zip" -o external/windeps.zip
-          7z x external/windeps.zip -oexternal/dlls
-
-      - name: Path to cached dependencies (Windows)
-        if: >
-          runner.os == 'Windows'
-        run: |
-          echo '${{ github.workspace }}'"/external/dlls" >> $GITHUB_PATH
-
-      - name: Derive environment variables
-        run: |
-          if [[ '${{ matrix.target.cpu }}' == 'amd64' ]]; then
-            PLATFORM=x64
-          else
-            PLATFORM=x86
-          fi
-          echo "PLATFORM=$PLATFORM" >> $GITHUB_ENV
-
-          ncpu=
-          MAKE_CMD="make"
-          case '${{ runner.os }}' in
-          'Linux')
-            ncpu=$(nproc)
-            ;;
-          'macOS')
-            ncpu=$(sysctl -n hw.ncpu)
-            ;;
-          'Windows')
-            ncpu=$NUMBER_OF_PROCESSORS
-            MAKE_CMD="mingw32-make"
-            ;;
-          esac
-          [[ -z "$ncpu" || $ncpu -le 0 ]] && ncpu=1
-          echo "ncpu=$ncpu" >> $GITHUB_ENV
-          echo "MAKE_CMD=${MAKE_CMD}" >> $GITHUB_ENV
-
-      - name: Build Nim and Nimble
-        run: |
-          curl -O -L -s -S https://raw.githubusercontent.com/status-im/nimbus-build-system/master/scripts/build_nim.sh
-          env MAKE="${MAKE_CMD} -j${ncpu}" ARCH_OVERRIDE=${PLATFORM} NIM_COMMIT=${{ matrix.branch }} \
-            QUICK_AND_DIRTY_COMPILER=1 QUICK_AND_DIRTY_NIMBLE=1 CC=gcc \
-            bash build_nim.sh nim csources dist/nimble NimBinaries
-          echo '${{ github.workspace }}/nim/bin' >> $GITHUB_PATH
+          os: ${{ matrix.target.os }}
+          shell: ${{ matrix.shell }}
+          nim_branch: ${{ matrix.branch }}
+          cpu: ${{ matrix.target.cpu }}
 
       - name: Setup Go
         uses: actions/setup-go@v2

.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

config.nims

@@ -1,6 +1,8 @@
 # to allow locking
 if dirExists("nimbledeps/pkgs"):
   switch("NimblePath", "nimbledeps/pkgs")
+if dirExists("nimbledeps/pkgs2"):
+  switch("NimblePath", "nimbledeps/pkgs2")
 
 switch("warning", "CaseTransition:off")
 switch("warning", "ObservableStores:off")

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/circuitrelay.nim

@@ -57,8 +57,7 @@ proc main() {.async.} =
   let
     # Create a relay address to swDst using swRel as the relay
     addrs = MultiAddress.init($swRel.peerInfo.addrs[0] & "/p2p/" &
-                              $swRel.peerInfo.peerId & "/p2p-circuit/p2p/" &
-                              $swDst.peerInfo.peerId).get()
+                              $swRel.peerInfo.peerId & "/p2p-circuit").get()
 
   # Connect Dst to the relay
   await swDst.connect(swRel.peerInfo.peerId, swRel.peerInfo.addrs)

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,7 +107,7 @@ 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()
@@ -115,8 +115,8 @@ proc new(_: typedesc[MetricProto], cb: MetricCallback): MetricProto =
     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.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p.nimble

@@ -22,9 +22,7 @@ requires "nim >= 1.2.0",
 import hashes
 proc runTest(filename: string, verify: bool = true, sign: bool = true,
              moreoptions: string = "") =
-  var excstr = "nim c --skipParentCfg --opt:speed -d:debug -d:libp2p_agents_metrics -d:libp2p_protobuf_metrics -d:libp2p_network_protocols_metrics -d:libp2p_mplex_metrics "
-  excstr.add(" -d:chronicles_sinks=textlines[stdout],json[dynamic] -d:chronicles_log_level=TRACE ")
-  excstr.add(" -d:chronicles_runtime_filtering=TRUE ")
+  var excstr = "nim c --skipParentCfg --opt:speed -d:debug "
   excstr.add(" " & getEnv("NIMFLAGS") & " ")
   excstr.add(" --verbosity:0 --hints:off ")
   excstr.add(" -d:libp2p_pubsub_sign=" & $sign)
@@ -139,10 +137,24 @@ 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:
-      rmDir(dependency)
+  let nimblePkgs =
+    if system.dirExists("nimbledeps/pkgs"): "nimbledeps/pkgs"
+    else: "nimbledeps/pkgs2"
+  for dependency in listDirs(nimblePkgs):
+    let
+      fileName = dependency.extractFilename
+      fileContent = readFile(dependency & "/nimblemeta.json")
+      packageName = fileName.split('-')[0]
+
+    if toInstall.anyIt(
+        it[0] == packageName and
+        (
+          it[1].split('#')[^1] in fileContent or # nimble for nim 2.X
+          fileName.endsWith(it[1].split('#')[^1]) # nimble for nim 1.X
+        )
+      ) == false or
+      fileName.split('-')[^1].len < 20: # safegard for nimble for nim 1.X
+        rmDir(dependency)
 
 task unpin, "Restore global package use":
   rmDir("nimbledeps")

libp2p/builders.nim

@@ -1,5 +1,5 @@
 # Nim-Libp2p
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))
@@ -27,7 +27,7 @@ import
   crypto/crypto, transports/[transport, tcptransport],
   muxers/[muxer, mplex/mplex, yamux/yamux],
   protocols/[identify, secure/secure, secure/noise, rendezvous],
-  protocols/connectivity/[autonat, relay/relay, relay/client, relay/rtransport],
+  protocols/connectivity/[autonat/server, relay/relay, relay/client, relay/rtransport],
   connmanager, upgrademngrs/muxedupgrade,
   nameresolving/nameresolver,
   errors, utility
@@ -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/cid.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/connmanager.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))
@@ -79,6 +79,7 @@ type
     muxed: Table[Connection, MuxerHolder]
     connEvents: array[ConnEventKind, OrderedSet[ConnEventHandler]]
     peerEvents: array[PeerEventKind, OrderedSet[PeerEventHandler]]
+    expectedConnections: Table[PeerId, Future[Connection]]
     peerStore*: PeerStore
 
   ConnectionSlot* = object
@@ -110,6 +111,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) =
@@ -213,6 +221,19 @@ proc triggerPeerEvents*(c: ConnManager,
   except CatchableError as exc: # handlers should not raise!
     warn "Exception in triggerPeerEvents", exc = exc.msg, peer = peerId
 
+proc expectConnection*(c: ConnManager, p: PeerId): Future[Connection] {.async.} =
+  ## Wait for a peer to connect to us. This will bypass the `MaxConnectionsPerPeer`
+  if p in c.expectedConnections:
+    raise LPError.newException("Already expecting a connection from that peer")
+
+  let future = newFuture[Connection]()
+  c.expectedConnections[p] = future
+
+  try:
+    return await future
+  finally:
+    c.expectedConnections.del(p)
+
 proc contains*(c: ConnManager, conn: Connection): bool =
   ## checks if a connection is being tracked by the
   ## connection manager
@@ -389,7 +410,12 @@ proc storeConn*(c: ConnManager, conn: Connection)
     raise newException(LPError, "Connection closed or EOF")
 
   let peerId = conn.peerId
-  if c.conns.getOrDefault(peerId).len > c.maxConnsPerPeer:
+
+  # we use getOrDefault in the if below instead of [] to avoid the KeyError
+  if peerId in c.expectedConnections and
+     not(c.expectedConnections.getOrDefault(peerId).finished):
+      c.expectedConnections.getOrDefault(peerId).complete(conn)
+  elif c.conns.getOrDefault(peerId).len > c.maxConnsPerPeer:
     debug "Too many connections for peer",
       conn, conns = c.conns.getOrDefault(peerId).len
 
@@ -409,7 +435,7 @@ proc getIncomingSlot*(c: ConnManager): Future[ConnectionSlot] {.async.} =
   await c.inSema.acquire()
   return ConnectionSlot(connManager: c, direction: In)
 
-proc getOutgoingSlot*(c: ConnManager, forceDial = false): Future[ConnectionSlot] {.async.} =
+proc getOutgoingSlot*(c: ConnManager, forceDial = false): ConnectionSlot {.raises: [Defect, TooManyConnectionsError].} =
   if forceDial:
     c.outSema.forceAcquire()
   elif not c.outSema.tryAcquire():
@@ -418,6 +444,13 @@ proc getOutgoingSlot*(c: ConnManager, forceDial = false): Future[ConnectionSlot]
     raise newTooManyConnectionsError()
   return ConnectionSlot(connManager: c, direction: Out)
 
+proc slotsAvailable*(c: ConnManager, dir: Direction): int =
+  case dir:
+    of Direction.In:
+      return c.inSema.count
+    of Direction.Out:
+      return c.outSema.count
+
 proc release*(cs: ConnectionSlot) =
   if cs.direction == In:
     cs.connManager.inSema.release()
@@ -529,6 +562,12 @@ proc close*(c: ConnManager) {.async.} =
   let muxed = c.muxed
   c.muxed.clear()
 
+  let expected = c.expectedConnections
+  c.expectedConnections.clear()
+
+  for _, fut in expected:
+    await fut.cancelAndWait()
+
   for _, muxer in muxed:
     await closeMuxerHolder(muxer)
 
@@ -537,3 +576,4 @@ proc close*(c: ConnManager) {.async.} =
       await conn.close()
 
   trace "Closed ConnManager"
+

libp2p/crypto/chacha20poly1305.nim

@@ -1,5 +1,5 @@
 # Nim-Libp2p
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/crypto/crypto.nim

@@ -1,5 +1,5 @@
 # Nim-Libp2p
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/crypto/curve25519.nim

@@ -1,5 +1,5 @@
 # Nim-Libp2p
-# Copyright (c) 2022-2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/crypto/ecnist.nim

@@ -1,5 +1,5 @@
 # Nim-Libp2p
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/crypto/ed25519/constants.nim

@@ -1,5 +1,5 @@
 # Nim-Libp2p
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/crypto/ed25519/ed25519.nim

@@ -1,5 +1,5 @@
 # Nim-Libp2p
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/crypto/hkdf.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/crypto/minasn1.nim

@@ -1,5 +1,5 @@
 # Nim-Libp2p
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/crypto/rsa.nim

@@ -1,5 +1,5 @@
 # Nim-Libp2p
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/crypto/secp.nim

@@ -1,5 +1,5 @@
 # Nim-Libp2p
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/daemon/daemonapi.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/daemon/transpool.nim

@@ -1,5 +1,5 @@
 # Nim-Libp2p
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/debugutils.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/dial.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))
@@ -27,7 +27,8 @@ method connect*(
   self: Dial,
   peerId: PeerId,
   addrs: seq[MultiAddress],
-  forceDial = false) {.async, base.} =
+  forceDial = false,
+  reuseConnection = true) {.async, base.} =
   ## connect remote peer without negotiating
   ## a protocol
   ##
@@ -36,7 +37,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

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))
@@ -60,7 +60,7 @@ proc dialAndUpgrade(
       let dialed =
         try:
           libp2p_total_dial_attempts.inc()
-          await transport.dial(hostname, address)
+          await transport.dial(hostname, address, peerId)
         except CancelledError as exc:
           debug "Dialing canceled", msg = exc.msg, peerId
           raise exc
@@ -147,11 +147,28 @@ proc dialAndUpgrade(
         if not isNil(result):
           return result
 
+proc tryReusingConnection(self: Dialer, peerId: PeerId): Future[Opt[Connection]] {.async.} =
+  var conn = self.connManager.selectConn(peerId)
+  if conn == nil:
+    return Opt.none(Connection)
+
+  if conn.atEof or conn.closed:
+    # This connection should already have been removed from the connection
+    # manager - it's essentially a bug that we end up here - we'll fail
+    # for now, hoping that this will clean themselves up later...
+    warn "dead connection in connection manager", conn
+    await conn.close()
+    raise newException(DialFailedError, "Zombie connection encountered")
+
+  trace "Reusing existing connection", conn, direction = $conn.dir
+  return Opt.some(conn)
+
 proc internalConnect(
   self: Dialer,
   peerId: Opt[PeerId],
   addrs: seq[MultiAddress],
-  forceDial: bool):
+  forceDial: bool,
+  reuseConnection = true):
   Future[Connection] {.async.} =
   if Opt.some(self.localPeerId) == peerId:
     raise newException(CatchableError, "can't dial self!")
@@ -161,24 +178,13 @@ proc internalConnect(
   try:
     await lock.acquire()
 
-    # Check if we have a connection already and try to reuse it
-    var conn =
-      if peerId.isSome: self.connManager.selectConn(peerId.get())
-      else: nil
-    if conn != nil:
-      if conn.atEof or conn.closed:
-        # This connection should already have been removed from the connection
-        # manager - it's essentially a bug that we end up here - we'll fail
-        # for now, hoping that this will clean themselves up later...
-        warn "dead connection in connection manager", conn
-        await conn.close()
-        raise newException(DialFailedError, "Zombie connection encountered")
+    if peerId.isSome and reuseConnection:
+      let connOpt = await self.tryReusingConnection(peerId.get())
+      if connOpt.isSome:
+        return connOpt.get()
 
-      trace "Reusing existing connection", conn, direction = $conn.dir
-      return conn
-
-    let slot = await self.connManager.getOutgoingSlot(forceDial)
-    conn =
+    let slot = self.connManager.getOutgoingSlot(forceDial)
+    let conn =
       try:
         await self.dialAndUpgrade(peerId, addrs)
       except CatchableError as exc:
@@ -207,23 +213,36 @@ method connect*(
   self: Dialer,
   peerId: PeerId,
   addrs: seq[MultiAddress],
-  forceDial = false) {.async.} =
+  forceDial = false,
+  reuseConnection = true) {.async.} =
   ## connect remote peer without negotiating
   ## a protocol
   ##
 
-  if self.connManager.connCount(peerId) > 0:
+  if self.connManager.connCount(peerId) > 0 and reuseConnection:
     return
 
-  discard await self.internalConnect(Opt.some(peerId), addrs, forceDial)
+  discard await self.internalConnect(Opt.some(peerId), addrs, forceDial, reuseConnection)
 
 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/discovery/discoverymngr.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/discovery/rendezvousinterface.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/multiaddress.nim

@@ -1,5 +1,5 @@
 # Nim-Libp2p
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))
@@ -462,6 +462,7 @@ const
   IP6* = mapEq("ip6")
   DNS* = mapOr(DNSANY, DNS4, DNS6, DNSADDR)
   IP* = mapOr(IP4, IP6)
+  DNS_OR_IP* = mapOr(DNS, IP)
   TCP* = mapOr(mapAnd(DNS, mapEq("tcp")), mapAnd(IP, mapEq("tcp")))
   UDP* = mapOr(mapAnd(DNS, mapEq("udp")), mapAnd(IP, mapEq("udp")))
   UTP* = mapAnd(UDP, mapEq("utp"))

libp2p/multibase.nim

@@ -1,5 +1,5 @@
 # Nim-Libp2p
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/multicodec.nim

@@ -1,5 +1,5 @@
 # Nim-Libp2p
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/multihash.nim

@@ -1,5 +1,5 @@
 # Nim-Libp2p
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/multistream.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))
@@ -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/mplex/coder.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/muxers/mplex/lpchannel.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))
@@ -64,6 +64,7 @@ type
     closeCode*: MessageType       # cached in/out close code
     resetCode*: MessageType       # cached in/out reset code
     writes*: int                  # In-flight writes
+    writesBytes*: int             # In-flight writes bytes
 
 func shortLog*(s: LPChannel): auto =
   try:
@@ -228,6 +229,7 @@ proc completeWrite(
     s: LPChannel, fut: Future[void], msgLen: int): Future[void] {.async.} =
   try:
     s.writes += 1
+    s.writesBytes += msgLen
 
     when defined(libp2p_mplex_metrics):
       libp2p_mplex_qlen.observe(s.writes.int64 - 1)
@@ -244,7 +246,11 @@ proc completeWrite(
   except CancelledError as exc:
     # Chronos may still send the data
     raise exc
-  except LPStreamClosedError as exc:
+  except LPStreamConnDownError as exc:
+    await s.reset()
+    await s.conn.close()
+    raise exc
+  except LPStreamEOFError as exc:
     raise exc
   except CatchableError as exc:
     trace "exception in lpchannel write handler", s, msg = exc.msg
@@ -253,6 +259,10 @@ proc completeWrite(
     raise newLPStreamConnDownError(exc)
   finally:
     s.writes -= 1
+    s.writesBytes -= msgLen
+
+
+method queuedSendBytes*(channel: LPChannel): int = channel.writesBytes
 
 method write*(s: LPChannel, msg: seq[byte]): Future[void] =
   ## Write to mplex channel - there may be up to MaxWrite concurrent writes

libp2p/muxers/mplex/mplex.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))
@@ -177,8 +177,14 @@ method handle*(m: Mplex) {.async, gcsafe.} =
             raise newLPStreamLimitError()
 
           trace "pushing data to channel", m, channel, len = data.len
-          await channel.pushData(data)
-          trace "pushed data to channel", m, channel, len = data.len
+          try:
+            await channel.pushData(data)
+            trace "pushed data to channel", m, channel, len = data.len
+          except LPStreamClosedError as exc:
+            # Channel is being closed, but `cleanupChann` was not yet triggered.
+            trace "pushing data to channel failed", m, channel, len = data.len,
+              msg = exc.msg
+            discard  # Ignore message, same as if `cleanupChann` had completed.
 
         of MessageType.CloseIn, MessageType.CloseOut:
           await channel.pushEof()

libp2p/muxers/muxer.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/muxers/yamux/yamux.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))
@@ -176,6 +176,8 @@ proc sendQueueBytes(channel: YamuxChannel, limit: bool = false): int =
   for (elem, sent, _) in channel.sendQueue:
     result.inc(min(elem.len - sent, if limit: channel.maxRecvWindow div 3 else: elem.len - sent))
 
+method queuedSendBytes*(channel: YamuxChannel): int = channel.sendQueueBytes()
+
 proc actuallyClose(channel: YamuxChannel) {.async.} =
   if channel.closedLocally and channel.sendQueue.len == 0 and
      channel.closedRemotely.done():
@@ -414,7 +416,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/nameresolving/dnsresolver.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/nameresolving/mockresolver.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/nameresolving/nameresolver.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/peerid.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))
@@ -184,6 +184,11 @@ func init*(t: typedesc[PeerId], seckey: PrivateKey): Result[PeerId, cstring] =
   ## Create new peer id from private key ``seckey``.
   PeerId.init(? seckey.getPublicKey().orError(cstring("invalid private key")))
 
+proc random*(t: typedesc[PeerId], rng = newRng()): Result[PeerId, cstring] =
+  ## Create new peer id with random public key.
+  let randomKey = PrivateKey.random(Secp256k1, rng[])[]
+  PeerId.init(randomKey).orError(cstring("failed to generate random key"))
+
 func match*(pid: PeerId, pubkey: PublicKey): bool =
   ## Returns ``true`` if ``pid`` matches public key ``pubkey``.
   let p = PeerId.init(pubkey)

libp2p/peerinfo.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/peerstore.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))
@@ -16,8 +16,7 @@ runnableExamples:
   # Create a custom book type
   type MoodBook = ref object of PeerBook[string]
 
-  var somePeerId: PeerId
-  discard somePeerId.init("")
+  var somePeerId = PeerId.random().get()
 
   peerStore[MoodBook][somePeerId] = "Happy"
   doAssert peerStore[MoodBook][somePeerId] == "Happy"
@@ -153,6 +152,9 @@ proc updatePeerInfo*(
   if info.addrs.len > 0:
     peerStore[AddressBook][info.peerId] = info.addrs
 
+  if info.pubkey.isSome:
+    peerStore[KeyBook][info.peerId] = info.pubkey.get()
+
   if info.agentVersion.isSome:
     peerStore[AgentBook][info.peerId] = info.agentVersion.get().string
 

libp2p/protobuf/minprotobuf.nim

@@ -1,5 +1,5 @@
 # Nim-Libp2p
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/protocols/connectivity/autonat.nim

@@ -1,309 +0,0 @@
-# Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
-# Licensed under either of
-#  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
-#  * MIT license ([LICENSE-MIT](LICENSE-MIT))
-# at your option.
-# This file may not be copied, modified, or distributed except according to
-# those terms.
-
-when (NimMajor, NimMinor) < (1, 4):
-  {.push raises: [Defect].}
-else:
-  {.push raises: [].}
-
-import std/[options, sets, sequtils]
-import stew/results
-import chronos, chronicles, stew/objects
-import ../protocol,
-       ../../switch,
-       ../../multiaddress,
-       ../../multicodec,
-       ../../peerid,
-       ../../utils/semaphore,
-       ../../errors
-
-logScope:
-  topics = "libp2p autonat"
-
-const
-  AutonatCodec* = "/libp2p/autonat/1.0.0"
-  AddressLimit = 8
-
-type
-  AutonatError* = object of LPError
-
-  MsgType* = enum
-    Dial = 0
-    DialResponse = 1
-
-  ResponseStatus* = enum
-    Ok = 0
-    DialError = 100
-    DialRefused = 101
-    BadRequest = 200
-    InternalError = 300
-
-  AutonatPeerInfo* = object
-    id: Option[PeerId]
-    addrs: seq[MultiAddress]
-
-  AutonatDial* = object
-    peerInfo: Option[AutonatPeerInfo]
-
-  AutonatDialResponse* = object
-    status*: ResponseStatus
-    text*: Option[string]
-    ma*: Option[MultiAddress]
-
-  AutonatMsg = object
-    msgType: MsgType
-    dial: Option[AutonatDial]
-    response: Option[AutonatDialResponse]
-
-proc encode*(msg: AutonatMsg): ProtoBuffer =
-  result = initProtoBuffer()
-  result.write(1, msg.msgType.uint)
-  if msg.dial.isSome():
-    var dial = initProtoBuffer()
-    if msg.dial.get().peerInfo.isSome():
-      var bufferPeerInfo = initProtoBuffer()
-      let peerInfo = msg.dial.get().peerInfo.get()
-      if peerInfo.id.isSome():
-        bufferPeerInfo.write(1, peerInfo.id.get())
-      for ma in peerInfo.addrs:
-        bufferPeerInfo.write(2, ma.data.buffer)
-      bufferPeerInfo.finish()
-      dial.write(1, bufferPeerInfo.buffer)
-    dial.finish()
-    result.write(2, dial.buffer)
-  if msg.response.isSome():
-    var bufferResponse = initProtoBuffer()
-    let response = msg.response.get()
-    bufferResponse.write(1, response.status.uint)
-    if response.text.isSome():
-      bufferResponse.write(2, response.text.get())
-    if response.ma.isSome():
-      bufferResponse.write(3, response.ma.get())
-    bufferResponse.finish()
-    result.write(3, bufferResponse.buffer)
-  result.finish()
-
-proc encode*(d: AutonatDial): ProtoBuffer =
-  result = initProtoBuffer()
-  result.write(1, MsgType.Dial.uint)
-  var dial = initProtoBuffer()
-  if d.peerInfo.isSome():
-    var bufferPeerInfo = initProtoBuffer()
-    let peerInfo = d.peerInfo.get()
-    if peerInfo.id.isSome():
-      bufferPeerInfo.write(1, peerInfo.id.get())
-    for ma in peerInfo.addrs:
-      bufferPeerInfo.write(2, ma.data.buffer)
-    bufferPeerInfo.finish()
-    dial.write(1, bufferPeerInfo.buffer)
-  dial.finish()
-  result.write(2, dial.buffer)
-  result.finish()
-
-proc encode*(r: AutonatDialResponse): ProtoBuffer =
-  result = initProtoBuffer()
-  result.write(1, MsgType.DialResponse.uint)
-  var bufferResponse = initProtoBuffer()
-  bufferResponse.write(1, r.status.uint)
-  if r.text.isSome():
-    bufferResponse.write(2, r.text.get())
-  if r.ma.isSome():
-    bufferResponse.write(3, r.ma.get())
-  bufferResponse.finish()
-  result.write(3, bufferResponse.buffer)
-  result.finish()
-
-proc decode(_: typedesc[AutonatMsg], buf: seq[byte]): Option[AutonatMsg] =
-  var
-    msgTypeOrd: uint32
-    pbDial: ProtoBuffer
-    pbResponse: ProtoBuffer
-    msg: AutonatMsg
-
-  let
-    pb = initProtoBuffer(buf)
-    r1 = pb.getField(1, msgTypeOrd)
-    r2 = pb.getField(2, pbDial)
-    r3 = pb.getField(3, pbResponse)
-  if r1.isErr() or r2.isErr() or r3.isErr(): return none(AutonatMsg)
-
-  if r1.get() and not checkedEnumAssign(msg.msgType, msgTypeOrd):
-    return none(AutonatMsg)
-  if r2.get():
-    var
-      pbPeerInfo: ProtoBuffer
-      dial: AutonatDial
-    let
-      r4 = pbDial.getField(1, pbPeerInfo)
-    if r4.isErr(): return none(AutonatMsg)
-
-    var peerInfo: AutonatPeerInfo
-    if r4.get():
-      var pid: PeerId
-      let
-        r5 = pbPeerInfo.getField(1, pid)
-        r6 = pbPeerInfo.getRepeatedField(2, peerInfo.addrs)
-      if r5.isErr() or r6.isErr(): return none(AutonatMsg)
-      if r5.get(): peerInfo.id = some(pid)
-      dial.peerInfo = some(peerInfo)
-    msg.dial = some(dial)
-
-  if r3.get():
-    var
-      statusOrd: uint
-      text: string
-      ma: MultiAddress
-      response: AutonatDialResponse
-
-    let
-      r4 = pbResponse.getField(1, statusOrd)
-      r5 = pbResponse.getField(2, text)
-      r6 = pbResponse.getField(3, ma)
-
-    if r4.isErr() or r5.isErr() or r6.isErr() or
-       (r4.get() and not checkedEnumAssign(response.status, statusOrd)):
-      return none(AutonatMsg)
-    if r5.get(): response.text = some(text)
-    if r6.get(): response.ma = some(ma)
-    msg.response = some(response)
-
-  return some(msg)
-
-proc sendDial(conn: Connection, pid: PeerId, addrs: seq[MultiAddress]) {.async.} =
-  let pb = AutonatDial(peerInfo: some(AutonatPeerInfo(
-                         id: some(pid),
-                         addrs: addrs
-                       ))).encode()
-  await conn.writeLp(pb.buffer)
-
-proc sendResponseError(conn: Connection, status: ResponseStatus, text: string = "") {.async.} =
-  let pb = AutonatDialResponse(
-             status: status,
-             text: if text == "": none(string) else: some(text),
-             ma: none(MultiAddress)
-           ).encode()
-  await conn.writeLp(pb.buffer)
-
-proc sendResponseOk(conn: Connection, ma: MultiAddress) {.async.} =
-  let pb = AutonatDialResponse(
-             status: ResponseStatus.Ok,
-             text: some("Ok"),
-             ma: some(ma)
-           ).encode()
-  await conn.writeLp(pb.buffer)
-
-type
-  Autonat* = ref object of LPProtocol
-    sem: AsyncSemaphore
-    switch*: Switch
-
-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()
-
-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)
-    if ma.isSome:
-      await conn.sendResponseOk(ma.get())
-    else:
-      await conn.sendResponseError(DialError, "Missing observed address")
-  except CancelledError as exc:
-    raise exc
-  except CatchableError as exc:
-    await conn.sendResponseError(DialError, exc.msg)
-  finally:
-    a.sem.release()
-
-proc handleDial(a: Autonat, conn: Connection, msg: AutonatMsg): Future[void] =
-  if msg.dial.isNone() or msg.dial.get().peerInfo.isNone():
-    return conn.sendResponseError(BadRequest, "Missing Peer Info")
-  let peerInfo = msg.dial.get().peerInfo.get()
-  if peerInfo.id.isSome() and peerInfo.id.get() != conn.peerId:
-    return conn.sendResponseError(BadRequest, "PeerId mismatch")
-
-  if conn.observedAddr.isNone:
-    return conn.sendResponseError(BadRequest, "Missing observed address")
-  let observedAddr = conn.observedAddr.get()
-
-  var isRelayed = observedAddr.contains(multiCodec("p2p-circuit"))
-  if isRelayed.isErr() or isRelayed.get():
-    return conn.sendResponseError(DialRefused, "Refused to dial a relayed observed address")
-  let hostIp = observedAddr[0]
-  if hostIp.isErr() or not IP.match(hostIp.get()):
-    trace "wrong observed address", address=observedAddr
-    return conn.sendResponseError(InternalError, "Expected an IP address")
-  var addrs = initHashSet[MultiAddress]()
-  addrs.incl(observedAddr)
-  for ma in peerInfo.addrs:
-    isRelayed = ma.contains(multiCodec("p2p-circuit"))
-    if isRelayed.isErr() or isRelayed.get():
-      continue
-    let maFirst = ma[0]
-    if maFirst.isErr() or not IP.match(maFirst.get()):
-      continue
-
-    try:
-      addrs.incl(
-        if maFirst.get() == hostIp.get():
-          ma
-        else:
-          let maEnd = ma[1..^1]
-          if maEnd.isErr(): continue
-          hostIp.get() & maEnd.get()
-      )
-    except LPError as exc:
-      continue
-    if len(addrs) >= AddressLimit:
-      break
-
-  if len(addrs) == 0:
-    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))
-  autonat.init()
-  autonat
-
-method init*(a: Autonat) =
-  proc handleStream(conn: Connection, proto: string) {.async, gcsafe.} =
-    try:
-      let msgOpt = AutonatMsg.decode(await conn.readLp(1024))
-      if msgOpt.isNone() or msgOpt.get().msgType != MsgType.Dial:
-        raise newException(AutonatError, "Received malformed message")
-      let msg = msgOpt.get()
-      await a.handleDial(conn, msg)
-    except CancelledError as exc:
-      raise exc
-    except CatchableError as exc:
-      trace "exception in autonat handler", exc = exc.msg, conn
-    finally:
-      trace "exiting autonat handler", conn
-      await conn.close()
-
-  a.handler = handleStream
-  a.codec = AutonatCodec

libp2p/protocols/connectivity/autonat/client.nim

@@ -0,0 +1,74 @@
+# Nim-LibP2P
+# Copyright (c) 2023 Status Research & Development GmbH
+# Licensed under either of
+#  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
+#  * MIT license ([LICENSE-MIT](LICENSE-MIT))
+# at your option.
+# This file may not be copied, modified, or distributed except according to
+# those terms.
+
+when (NimMajor, NimMinor) < (1, 4):
+  {.push raises: [Defect].}
+else:
+  {.push raises: [].}
+
+import std/[options, sets, sequtils]
+import stew/[results, objects]
+import chronos, chronicles
+import ../../../switch,
+       ../../../multiaddress,
+       ../../../peerid
+import core
+
+export core
+
+logScope:
+  topics = "libp2p autonat"
+
+type
+  AutonatClient* = ref object of RootObj
+
+proc sendDial(conn: Connection, pid: PeerId, addrs: seq[MultiAddress]) {.async.} =
+  let pb = AutonatDial(peerInfo: some(AutonatPeerInfo(
+                         id: some(pid),
+                         addrs: addrs
+                       ))).encode()
+  await conn.writeLp(pb.buffer)
+
+method dialMe*(self: AutonatClient, switch: Switch, 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 switch.dial(pid, @[AutonatCodec])
+      else:
+        await switch.dial(pid, addrs, AutonatCodec)
+    except CatchableError as err:
+      raise newException(AutonatError, "Unexpected error when dialling: " & err.msg, err)
+
+  # To bypass maxConnectionsPerPeer
+  let incomingConnection = switch.connManager.expectConnection(pid)
+  defer:
+    await conn.close()
+    incomingConnection.cancel() # Safer to always try to cancel cause we aren't sure if the peer dialled us or not
+  trace "sending Dial", addrs = switch.peerInfo.addrs
+  await conn.sendDial(switch.peerInfo.peerId, switch.peerInfo.addrs)
+  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: " & response.text.get(""))
+    else:
+      raise newException(AutonatError, "Bad status " & $response.status & " " & response.text.get(""))

libp2p/protocols/connectivity/autonat/core.nim

@@ -0,0 +1,152 @@
+# Nim-LibP2P
+# Copyright (c) 2023 Status Research & Development GmbH
+# Licensed under either of
+#  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
+#  * MIT license ([LICENSE-MIT](LICENSE-MIT))
+# at your option.
+# This file may not be copied, modified, or distributed except according to
+# those terms.
+
+when (NimMajor, NimMinor) < (1, 4):
+  {.push raises: [Defect].}
+else:
+  {.push raises: [].}
+
+import std/[options, sets, sequtils]
+import stew/[results, objects]
+import chronos, chronicles
+import ../../../multiaddress,
+       ../../../peerid,
+       ../../../errors
+
+logScope:
+  topics = "libp2p autonat"
+
+const
+  AutonatCodec* = "/libp2p/autonat/1.0.0"
+  AddressLimit* = 8
+
+type
+  AutonatError* = object of LPError
+  AutonatUnreachableError* = object of LPError
+
+  MsgType* = enum
+    Dial = 0
+    DialResponse = 1
+
+  ResponseStatus* = enum
+    Ok = 0
+    DialError = 100
+    DialRefused = 101
+    BadRequest = 200
+    InternalError = 300
+
+  AutonatPeerInfo* = object
+    id*: Option[PeerId]
+    addrs*: seq[MultiAddress]
+
+  AutonatDial* = object
+    peerInfo*: Option[AutonatPeerInfo]
+
+  AutonatDialResponse* = object
+    status*: ResponseStatus
+    text*: Option[string]
+    ma*: Option[MultiAddress]
+
+  AutonatMsg* = object
+    msgType*: MsgType
+    dial*: Option[AutonatDial]
+    response*: Option[AutonatDialResponse]
+
+proc encode(p: AutonatPeerInfo): ProtoBuffer =
+  result = initProtoBuffer()
+  if p.id.isSome():
+    result.write(1, p.id.get())
+  for ma in p.addrs:
+    result.write(2, ma.data.buffer)
+  result.finish()
+
+proc encode*(d: AutonatDial): ProtoBuffer =
+  result = initProtoBuffer()
+  result.write(1, MsgType.Dial.uint)
+  var dial = initProtoBuffer()
+  if d.peerInfo.isSome():
+    dial.write(1, encode(d.peerInfo.get()))
+  dial.finish()
+  result.write(2, dial.buffer)
+  result.finish()
+
+proc encode*(r: AutonatDialResponse): ProtoBuffer =
+  result = initProtoBuffer()
+  result.write(1, MsgType.DialResponse.uint)
+  var bufferResponse = initProtoBuffer()
+  bufferResponse.write(1, r.status.uint)
+  if r.text.isSome():
+    bufferResponse.write(2, r.text.get())
+  if r.ma.isSome():
+    bufferResponse.write(3, r.ma.get())
+  bufferResponse.finish()
+  result.write(3, bufferResponse.buffer)
+  result.finish()
+
+proc encode*(msg: AutonatMsg): ProtoBuffer =
+  if msg.dial.isSome():
+    return encode(msg.dial.get())
+  if msg.response.isSome():
+    return encode(msg.response.get())
+
+proc decode*(_: typedesc[AutonatMsg], buf: seq[byte]): Option[AutonatMsg] =
+  var
+    msgTypeOrd: uint32
+    pbDial: ProtoBuffer
+    pbResponse: ProtoBuffer
+    msg: AutonatMsg
+
+  let
+    pb = initProtoBuffer(buf)
+    r1 = pb.getField(1, msgTypeOrd)
+    r2 = pb.getField(2, pbDial)
+    r3 = pb.getField(3, pbResponse)
+  if r1.isErr() or r2.isErr() or r3.isErr(): return none(AutonatMsg)
+
+  if r1.get() and not checkedEnumAssign(msg.msgType, msgTypeOrd):
+    return none(AutonatMsg)
+  if r2.get():
+    var
+      pbPeerInfo: ProtoBuffer
+      dial: AutonatDial
+    let
+      r4 = pbDial.getField(1, pbPeerInfo)
+    if r4.isErr(): return none(AutonatMsg)
+
+    var peerInfo: AutonatPeerInfo
+    if r4.get():
+      var pid: PeerId
+      let
+        r5 = pbPeerInfo.getField(1, pid)
+        r6 = pbPeerInfo.getRepeatedField(2, peerInfo.addrs)
+      if r5.isErr() or r6.isErr(): return none(AutonatMsg)
+      if r5.get(): peerInfo.id = some(pid)
+      dial.peerInfo = some(peerInfo)
+    msg.dial = some(dial)
+
+  if r3.get():
+    var
+      statusOrd: uint
+      text: string
+      ma: MultiAddress
+      response: AutonatDialResponse
+
+    let
+      r4 = pbResponse.getField(1, statusOrd)
+      r5 = pbResponse.getField(2, text)
+      r6 = pbResponse.getField(3, ma)
+
+    if r4.isErr() or r5.isErr() or r6.isErr() or
+       (r4.get() and not checkedEnumAssign(response.status, statusOrd)):
+      return none(AutonatMsg)
+    if r5.get(): response.text = some(text)
+    if r6.get(): response.ma = some(ma)
+    msg.response = some(response)
+
+  return some(msg)

libp2p/protocols/connectivity/autonat/server.nim

@@ -0,0 +1,162 @@
+# Nim-LibP2P
+# Copyright (c) 2023 Status Research & Development GmbH
+# Licensed under either of
+#  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
+#  * MIT license ([LICENSE-MIT](LICENSE-MIT))
+# at your option.
+# This file may not be copied, modified, or distributed except according to
+# those terms.
+
+when (NimMajor, NimMinor) < (1, 4):
+  {.push raises: [Defect].}
+else:
+  {.push raises: [].}
+
+import std/[options, sets, sequtils]
+import stew/results
+import chronos, chronicles, stew/objects
+import ../../protocol,
+       ../../../switch,
+       ../../../multiaddress,
+       ../../../multicodec,
+       ../../../peerid,
+       ../../../utils/[semaphore, future],
+       ../../../errors
+import core
+
+export core
+
+logScope:
+  topics = "libp2p autonat"
+
+type
+  Autonat* = ref object of LPProtocol
+    sem: AsyncSemaphore
+    switch*: Switch
+    dialTimeout: Duration
+
+proc sendDial(conn: Connection, pid: PeerId, addrs: seq[MultiAddress]) {.async.} =
+  let pb = AutonatDial(peerInfo: some(AutonatPeerInfo(
+                         id: some(pid),
+                         addrs: addrs
+                       ))).encode()
+  await conn.writeLp(pb.buffer)
+
+proc sendResponseError(conn: Connection, status: ResponseStatus, text: string = "") {.async.} =
+  let pb = AutonatDialResponse(
+             status: status,
+             text: if text == "": none(string) else: some(text),
+             ma: none(MultiAddress)
+           ).encode()
+  await conn.writeLp(pb.buffer)
+
+proc sendResponseOk(conn: Connection, ma: MultiAddress) {.async.} =
+  let pb = AutonatDialResponse(
+             status: ResponseStatus.Ok,
+             text: some("Ok"),
+             ma: some(ma)
+           ).encode()
+  await conn.writeLp(pb.buffer)
+
+proc tryDial(autonat: Autonat, conn: Connection, addrs: seq[MultiAddress]) {.async.} =
+  await autonat.sem.acquire()
+  var futs: seq[Future[Opt[MultiAddress]]]
+  try:
+    # This is to bypass the per peer max connections limit
+    let outgoingConnection = autonat.switch.connManager.expectConnection(conn.peerId)
+    # Safer to always try to cancel cause we aren't sure if the connection was established
+    defer: outgoingConnection.cancel()
+    # tryDial is to bypass the global max connections limit
+    futs = addrs.mapIt(autonat.switch.dialer.tryDial(conn.peerId, @[it]))
+    let fut = await anyCompleted(futs).wait(autonat.dialTimeout)
+    let ma = await fut
+    if ma.isSome:
+      await conn.sendResponseOk(ma.get())
+    else:
+      await conn.sendResponseError(DialError, "Missing observed address")
+  except CancelledError as exc:
+    raise exc
+  except AllFuturesFailedError as exc:
+    debug "All dial attempts failed", addrs, exc = exc.msg
+    await conn.sendResponseError(DialError, "All dial attempts failed")
+  except AsyncTimeoutError as exc:
+    debug "Dial timeout", addrs, exc = exc.msg
+    await conn.sendResponseError(DialError, "Dial timeout")
+  except CatchableError as exc:
+    debug "Unexpected error", addrs, exc = exc.msg
+    await conn.sendResponseError(DialError, "Unexpected error")
+  finally:
+    autonat.sem.release()
+    for f in futs:
+      if not f.finished():
+        f.cancel()
+
+proc handleDial(autonat: Autonat, conn: Connection, msg: AutonatMsg): Future[void] =
+  if msg.dial.isNone() or msg.dial.get().peerInfo.isNone():
+    return conn.sendResponseError(BadRequest, "Missing Peer Info")
+  let peerInfo = msg.dial.get().peerInfo.get()
+  if peerInfo.id.isSome() and peerInfo.id.get() != conn.peerId:
+    return conn.sendResponseError(BadRequest, "PeerId mismatch")
+
+  if conn.observedAddr.isNone:
+    return conn.sendResponseError(BadRequest, "Missing observed address")
+  let observedAddr = conn.observedAddr.get()
+
+  var isRelayed = observedAddr.contains(multiCodec("p2p-circuit"))
+  if isRelayed.isErr() or isRelayed.get():
+    return conn.sendResponseError(DialRefused, "Refused to dial a relayed observed address")
+  let hostIp = observedAddr[0]
+  if hostIp.isErr() or not IP.match(hostIp.get()):
+    trace "wrong observed address", address=observedAddr
+    return conn.sendResponseError(InternalError, "Expected an IP address")
+  var addrs = initHashSet[MultiAddress]()
+  addrs.incl(observedAddr)
+  trace "addrs received", addrs = peerInfo.addrs
+  for ma in peerInfo.addrs:
+    isRelayed = ma.contains(multiCodec("p2p-circuit"))
+    if isRelayed.isErr() or isRelayed.get():
+      continue
+    let maFirst = ma[0]
+    if maFirst.isErr() or not DNS_OR_IP.match(maFirst.get()):
+      continue
+
+    try:
+      addrs.incl(
+        if maFirst.get() == hostIp.get():
+          ma
+        else:
+          let maEnd = ma[1..^1]
+          if maEnd.isErr(): continue
+          hostIp.get() & maEnd.get()
+      )
+    except LPError as exc:
+      continue
+    if len(addrs) >= AddressLimit:
+      break
+
+  if len(addrs) == 0:
+    return conn.sendResponseError(DialRefused, "No dialable address")
+  let addrsSeq = toSeq(addrs)
+  trace "trying to dial", addrs = addrsSeq
+  return autonat.tryDial(conn, addrsSeq)
+
+proc new*(T: typedesc[Autonat], switch: Switch, semSize: int = 1, dialTimeout = 15.seconds): T =
+  let autonat = T(switch: switch, sem: newAsyncSemaphore(semSize), dialTimeout: dialTimeout)
+  proc handleStream(conn: Connection, proto: string) {.async, gcsafe.} =
+    try:
+      let msgOpt = AutonatMsg.decode(await conn.readLp(1024))
+      if msgOpt.isNone() or msgOpt.get().msgType != MsgType.Dial:
+        raise newException(AutonatError, "Received malformed message")
+      let msg = msgOpt.get()
+      await autonat.handleDial(conn, msg)
+    except CancelledError as exc:
+      raise exc
+    except CatchableError as exc:
+      debug "exception in autonat handler", exc = exc.msg, conn
+    finally:
+      trace "exiting autonat handler", conn
+      await conn.close()
+
+  autonat.handler = handleStream
+  autonat.codec = AutonatCodec
+  autonat

libp2p/protocols/connectivity/autonat/service.nim

@@ -0,0 +1,182 @@
+# Nim-LibP2P
+# Copyright (c) 2023 Status Research & Development GmbH
+# Licensed under either of
+#  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
+#  * MIT license ([LICENSE-MIT](LICENSE-MIT))
+# at your option.
+# This file may not be copied, modified, or distributed except according to
+# those terms.
+
+when (NimMajor, NimMinor) < (1, 4):
+  {.push raises: [Defect].}
+else:
+  {.push raises: [].}
+
+import std/[options, deques, sequtils]
+import chronos, metrics
+import ../../../switch
+import client
+import ../../../utils/heartbeat
+import ../../../crypto/crypto
+
+logScope:
+  topics = "libp2p autonatservice"
+
+declarePublicGauge(libp2p_autonat_reachability_confidence, "autonat reachability confidence", labels = ["reachability"])
+
+type
+  AutonatService* = ref object of Service
+    newConnectedPeerHandler: PeerEventHandler
+    scheduleHandle: Future[void]
+    networkReachability: NetworkReachability
+    confidence: Option[float]
+    answers: Deque[NetworkReachability]
+    autonatClient: AutonatClient
+    statusAndConfidenceHandler: StatusAndConfidenceHandler
+    rng: ref HmacDrbgContext
+    scheduleInterval: Option[Duration]
+    askNewConnectedPeers: bool
+    numPeersToAsk: int
+    maxQueueSize: int
+    minConfidence: float
+    dialTimeout: Duration
+
+  NetworkReachability* {.pure.} = enum
+    NotReachable, Reachable, Unknown
+
+  StatusAndConfidenceHandler* = proc (networkReachability: NetworkReachability, confidence: Option[float]): Future[void]  {.gcsafe, raises: [Defect].}
+
+proc new*(
+  T: typedesc[AutonatService],
+  autonatClient: AutonatClient,
+  rng: ref HmacDrbgContext,
+  scheduleInterval: Option[Duration] = none(Duration),
+  askNewConnectedPeers = true,
+  numPeersToAsk: int = 5,
+  maxQueueSize: int = 10,
+  minConfidence: float = 0.3,
+  dialTimeout = 30.seconds): T =
+  return T(
+    scheduleInterval: scheduleInterval,
+    networkReachability: Unknown,
+    confidence: none(float),
+    answers: initDeque[NetworkReachability](),
+    autonatClient: autonatClient,
+    rng: rng,
+    askNewConnectedPeers: askNewConnectedPeers,
+    numPeersToAsk: numPeersToAsk,
+    maxQueueSize: maxQueueSize,
+    minConfidence: minConfidence,
+    dialTimeout: dialTimeout)
+
+proc networkReachability*(self: AutonatService): NetworkReachability {.inline.} =
+  return self.networkReachability
+
+proc callHandler(self: AutonatService) {.async.} =
+  if not isNil(self.statusAndConfidenceHandler):
+    await self.statusAndConfidenceHandler(self.networkReachability, self.confidence)
+
+proc hasEnoughIncomingSlots(switch: Switch): bool =
+  # we leave some margin instead of comparing to 0 as a peer could connect to us while we are asking for the dial back
+  return switch.connManager.slotsAvailable(In) >= 2
+
+proc handleAnswer(self: AutonatService, ans: NetworkReachability) {.async.} =
+
+  if ans == Unknown:
+    return
+
+  if self.answers.len == self.maxQueueSize:
+    self.answers.popFirst()
+  self.answers.addLast(ans)
+
+  self.networkReachability = Unknown
+  self.confidence = none(float)
+  const reachabilityPriority = [Reachable, NotReachable]
+  for reachability in reachabilityPriority:
+    let confidence = self.answers.countIt(it == reachability) / self.maxQueueSize
+    libp2p_autonat_reachability_confidence.set(value = confidence, labelValues = [$reachability])
+    if self.confidence.isNone and confidence >= self.minConfidence:
+      self.networkReachability = reachability
+      self.confidence = some(confidence)
+
+  debug "Current status", currentStats = $self.networkReachability, confidence = $self.confidence, answers = self.answers
+
+proc askPeer(self: AutonatService, switch: Switch, peerId: PeerId): Future[NetworkReachability] {.async.} =
+  logScope:
+    peerId = $peerId
+  if not hasEnoughIncomingSlots(switch):
+    debug "No incoming slots available, not asking peer", incomingSlotsAvailable=switch.connManager.slotsAvailable(In)
+    return Unknown
+
+  trace "Asking peer for reachability"
+  let ans =
+    try:
+      discard await self.autonatClient.dialMe(switch, peerId).wait(self.dialTimeout)
+      debug "dialMe answer is reachable"
+      Reachable
+    except AutonatUnreachableError as error:
+      debug "dialMe answer is not reachable", msg = error.msg
+      NotReachable
+    except AsyncTimeoutError as error:
+      debug "dialMe timed out", msg = error.msg
+      Unknown
+    except CatchableError as error:
+      debug "dialMe unexpected error", msg = error.msg
+      Unknown
+  await self.handleAnswer(ans)
+  if not isNil(self.statusAndConfidenceHandler):
+    await self.statusAndConfidenceHandler(self.networkReachability, self.confidence)
+  return ans
+
+proc askConnectedPeers(self: AutonatService, switch: Switch) {.async.} =
+  trace "Asking peers for reachability"
+  var peers = switch.connectedPeers(Direction.Out)
+  self.rng.shuffle(peers)
+  var answersFromPeers = 0
+  for peer in peers:
+    if answersFromPeers >= self.numPeersToAsk:
+      break
+    if not hasEnoughIncomingSlots(switch):
+      debug "No incoming slots available, not asking peers", incomingSlotsAvailable=switch.connManager.slotsAvailable(In)
+      break
+    if (await askPeer(self, switch, peer)) != Unknown:
+      answersFromPeers.inc()
+
+proc schedule(service: AutonatService, switch: Switch, interval: Duration) {.async.} =
+  heartbeat "Scheduling AutonatService run", interval:
+    await service.run(switch)
+
+method setup*(self: AutonatService, switch: Switch): Future[bool] {.async.} =
+  info "Setting up AutonatService"
+  let hasBeenSetup = await procCall Service(self).setup(switch)
+  if hasBeenSetup:
+    if self.askNewConnectedPeers:
+      self.newConnectedPeerHandler = proc (peerId: PeerId, event: PeerEvent): Future[void] {.async.} =
+        if switch.connManager.selectConn(peerId, In) != nil: # no need to ask an incoming peer
+          return
+        discard askPeer(self, switch, peerId)
+        await self.callHandler()
+      switch.connManager.addPeerEventHandler(self.newConnectedPeerHandler, PeerEventKind.Joined)
+    if self.scheduleInterval.isSome():
+      self.scheduleHandle = schedule(self, switch, self.scheduleInterval.get())
+  return hasBeenSetup
+
+method run*(self: AutonatService, switch: Switch) {.async, public.} =
+  trace "Running AutonatService"
+  await askConnectedPeers(self, switch)
+  await self.callHandler()
+
+
+method stop*(self: AutonatService, switch: Switch): Future[bool] {.async, public.} =
+  info "Stopping AutonatService"
+  let hasBeenStopped = await procCall Service(self).stop(switch)
+  if hasBeenStopped:
+    if not isNil(self.scheduleHandle):
+      self.scheduleHandle.cancel()
+      self.scheduleHandle = nil
+    if not isNil(self.newConnectedPeerHandler):
+      switch.connManager.removePeerEventHandler(self.newConnectedPeerHandler, PeerEventKind.Joined)
+  return hasBeenStopped
+
+proc statusAndConfidenceHandler*(self: AutonatService, statusAndConfidenceHandler: StatusAndConfidenceHandler) =
+  self.statusAndConfidenceHandler = statusAndConfidenceHandler

libp2p/protocols/connectivity/relay/client.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))
@@ -25,6 +25,7 @@ import ./relay,
        ../../../multiaddress,
        ../../../stream/connection
 
+export options
 
 logScope:
   topics = "libp2p relay relay-client"

libp2p/protocols/connectivity/relay/messages.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/protocols/connectivity/relay/rconn.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/protocols/connectivity/relay/relay.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/protocols/connectivity/relay/rtransport.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))
@@ -92,15 +92,15 @@ proc dial*(self: RelayTransport, ma: MultiAddress): Future[Connection] {.async,
 method dial*(
   self: RelayTransport,
   hostname: string,
-  address: MultiAddress): Future[Connection] {.async, gcsafe.} =
+  ma: MultiAddress,
+  peerId: Opt[PeerId] = Opt.none(PeerId)): Future[Connection] {.async, gcsafe.} =
+  let address = MultiAddress.init($ma & "/p2p/" & $peerId.get()).tryGet()
   result = await self.dial(address)
 
 method handles*(self: RelayTransport, ma: MultiAddress): bool {.gcsafe} =
   if ma.protocols.isOk():
     let sma = toSeq(ma.items())
-    if sma.len >= 3:
-      result = CircuitRelay.match(sma[^2].get()) and
-               P2PPattern.match(sma[^1].get())
+    result = sma.len >= 2 and CircuitRelay.match(sma[^1].get())
   trace "Handles return", ma, result
 
 proc new*(T: typedesc[RelayTransport], cl: RelayClient, upgrader: Upgrade): T =

libp2p/protocols/connectivity/relay/utils.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/protocols/identify.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/protocols/ping.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/protocols/protocol.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))
@@ -12,9 +12,14 @@ when (NimMajor, NimMinor) < (1, 4):
 else:
   {.push raises: [].}
 
-import chronos
+import chronos, stew/results
 import ../stream/connection
 
+export results
+
+const
+  DefaultMaxIncomingStreams* = 10
+
 type
   LPProtoHandler* = proc (
     conn: Connection,
@@ -26,11 +31,17 @@ type
     codecs*: seq[string]
     handler*: LPProtoHandler ## this handler gets invoked by the protocol negotiator
     started*: bool
+    maxIncomingStreams: Opt[int]
 
 method init*(p: LPProtocol) {.base, gcsafe.} = discard
 method start*(p: LPProtocol) {.async, base.} = p.started = true
 method stop*(p: LPProtocol) {.async, base.} = p.started = false
 
+proc maxIncomingStreams*(p: LPProtocol): int =
+  p.maxIncomingStreams.get(DefaultMaxIncomingStreams)
+
+proc `maxIncomingStreams=`*(p: LPProtocol, val: int) =
+  p.maxIncomingStreams = Opt.some(val)
 
 func codec*(p: LPProtocol): string =
   assert(p.codecs.len > 0, "Codecs sequence was empty!")
@@ -40,3 +51,16 @@ func `codec=`*(p: LPProtocol, codec: string) =
   # always insert as first codec
   # if we use this abstraction
   p.codecs.insert(codec, 0)
+
+proc new*(
+  T: type LPProtocol,
+  codecs: seq[string],
+  handler: LPProtoHandler,  # default(Opt[int]) or Opt.none(int) don't work on 1.2
+  maxIncomingStreams: Opt[int] | int = Opt[int]()): T =
+  T(
+    codecs: codecs,
+    handler: handler,
+    maxIncomingStreams:
+      when maxIncomingStreams is int: Opt.some(maxIncomingStreams)
+      else: maxIncomingStreams
+  )

libp2p/protocols/pubsub/floodsub.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/protocols/pubsub/gossipsub.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/protocols/pubsub/gossipsub/behavior.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/protocols/pubsub/gossipsub/scoring.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/protocols/pubsub/gossipsub/types.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/protocols/pubsub/mcache.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/protocols/pubsub/peertable.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/protocols/pubsub/pubsub.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))
@@ -202,7 +202,7 @@ proc broadcast*(
     # Fast path that only encodes message once
     let encoded = encodeRpcMsg(msg, p.anonymize)
     for peer in sendPeers:
-      peer.sendEncoded(encoded)
+      asyncSpawn peer.sendEncoded(encoded)
 
 proc sendSubs*(p: PubSub,
                peer: PubSubPeer,
@@ -307,8 +307,6 @@ proc getOrCreatePeer*(
   # metrics
   libp2p_pubsub_peers.set(p.peers.len.int64)
 
-  pubSubPeer.connect()
-
   return pubSubPeer
 
 proc handleData*(p: PubSub, topic: string, data: seq[byte]): Future[void] =
@@ -382,7 +380,8 @@ method subscribePeer*(p: PubSub, peer: PeerId) {.base, gcsafe.} =
   ## messages
   ##
 
-  discard p.getOrCreatePeer(peer, p.codecs)
+  let pubSubPeer = p.getOrCreatePeer(peer, p.codecs)
+  pubSubPeer.connect()
 
 proc updateTopicMetrics(p: PubSub, topic: string) =
   # metrics

libp2p/protocols/pubsub/pubsubpeer.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))
@@ -55,6 +55,7 @@ type
     onEvent*: OnEvent                   # Connectivity updates for peer
     codec*: string                      # the protocol that this peer joined from
     sendConn*: Connection               # cached send connection
+    connectedFut: Future[void]
     address*: Option[MultiAddress]
     peerId*: PeerId
     handler*: RPCHandler
@@ -73,6 +74,10 @@ type
   RPCHandler* = proc(peer: PubSubPeer, msg: RPCMsg): Future[void]
     {.gcsafe, raises: [Defect].}
 
+proc queuedSendBytes*(p: PubSubPeer): int =
+  if p.sendConn.isNil: 0
+  else: p.sendConn.queuedSendBytes()
+
 func hash*(p: PubSubPeer): Hash =
   p.peerId.hash
 
@@ -165,6 +170,8 @@ proc handle*(p: PubSubPeer, conn: Connection) {.async.} =
 
 proc connectOnce(p: PubSubPeer): Future[void] {.async.} =
   try:
+    if p.connectedFut.finished:
+      p.connectedFut = newFuture[void]()
     let newConn = await p.getConn()
     if newConn.isNil:
       raise (ref LPError)(msg: "Cannot establish send connection")
@@ -174,6 +181,7 @@ proc connectOnce(p: PubSubPeer): Future[void] {.async.} =
     # stop working so we make an effort to only keep a single channel alive
 
     trace "Get new send connection", p, newConn
+    p.connectedFut.complete()
     p.sendConn = newConn
     p.address = if p.sendConn.observedAddr.isSome: some(p.sendConn.observedAddr.get) else: none(MultiAddress)
 
@@ -208,28 +216,11 @@ proc connectImpl(p: PubSubPeer) {.async.} =
     debug "Could not establish send connection", msg = exc.msg
 
 proc connect*(p: PubSubPeer) =
+  if p.connected:
+    return
+
   asyncSpawn connectImpl(p)
 
-proc sendImpl(conn: Connection, encoded: seq[byte]): Future[void] {.raises: [Defect].} =
-  trace "sending encoded msgs to peer", conn, encoded = shortLog(encoded)
-
-  let fut = conn.writeLp(encoded) # Avoid copying `encoded` into future
-  proc sendWaiter(): Future[void] {.async.} =
-    try:
-      await fut
-      trace "sent pubsub message to remote", conn
-
-    except CatchableError as exc: # never cancelled
-      # Because we detach the send call from the currently executing task using
-      # asyncSpawn, no exceptions may leak out of it
-      trace "Unable to send to remote", conn, msg = exc.msg
-      # Next time sendConn is used, it will be have its close flag set and thus
-      # will be recycled
-
-      await conn.close() # This will clean up the send connection
-
-  return sendWaiter()
-
 template sendMetrics(msg: RPCMsg): untyped =
   when defined(libp2p_expensive_metrics):
     for x in msg.messages:
@@ -237,7 +228,7 @@ template sendMetrics(msg: RPCMsg): untyped =
         # metrics
         libp2p_pubsub_sent_messages.inc(labelValues = [$p.peerId, t])
 
-proc sendEncoded*(p: PubSubPeer, msg: seq[byte]) {.raises: [Defect].} =
+proc sendEncoded*(p: PubSubPeer, msg: seq[byte]) {.raises: [Defect], async.} =
   doAssert(not isNil(p), "pubsubpeer nil!")
 
   if msg.len <= 0:
@@ -248,14 +239,27 @@ proc sendEncoded*(p: PubSubPeer, msg: seq[byte]) {.raises: [Defect].} =
     info "trying to send a too big for pubsub", maxSize=p.maxMessageSize, msgSize=msg.len
     return
 
-  let conn = p.sendConn
+  if p.sendConn == nil:
+    discard await p.connectedFut.withTimeout(1.seconds)
+
+  var conn = p.sendConn
   if conn == nil or conn.closed():
-    trace "No send connection, skipping message", p, msg = shortLog(msg)
+    debug "No send connection, skipping message", p, msg = shortLog(msg)
     return
 
-  # To limit the size of the closure, we only pass the encoded message and
-  # connection to the spawned send task
-  asyncSpawn sendImpl(conn, msg)
+  trace "sending encoded msgs to peer", conn, encoded = shortLog(msg)
+
+  try:
+    await conn.writeLp(msg)
+    trace "sent pubsub message to remote", conn
+  except CatchableError as exc: # never cancelled
+    # Because we detach the send call from the currently executing task using
+    # asyncSpawn, no exceptions may leak out of it
+    trace "Unable to send to remote", conn, msg = exc.msg
+    # Next time sendConn is used, it will be have its close flag set and thus
+    # will be recycled
+
+    await conn.close() # This will clean up the send connection
 
 proc send*(p: PubSubPeer, msg: RPCMsg, anonymize: bool) {.raises: [Defect].} =
   trace "sending msg to peer", peer = p, rpcMsg = shortLog(msg)
@@ -277,7 +281,7 @@ proc send*(p: PubSubPeer, msg: RPCMsg, anonymize: bool) {.raises: [Defect].} =
     sendMetrics(msg)
     encodeRpcMsg(msg, anonymize)
 
-  p.sendEncoded(encoded)
+  asyncSpawn p.sendEncoded(encoded)
 
 proc new*(
   T: typedesc[PubSubPeer],
@@ -292,5 +296,6 @@ proc new*(
     onEvent: onEvent,
     codec: codec,
     peerId: peerId,
+    connectedFut: newFuture[void](),
     maxMessageSize: maxMessageSize
   )

libp2p/protocols/pubsub/rpc/message.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/protocols/pubsub/rpc/messages.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/protocols/pubsub/rpc/protobuf.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))
@@ -304,14 +304,15 @@ proc decodeMessages*(pb: ProtoBuffer): ProtoResult[seq[Message]] {.inline.} =
   if ? pb.getRepeatedField(2, msgpbs):
     trace "decodeMessages: read messages", count = len(msgpbs)
     for item in msgpbs:
-      msgs.add(? decodeMessage(initProtoBuffer(item)))
+      # size is constrained at the network level
+      msgs.add(? decodeMessage(initProtoBuffer(item, maxSize = uint.high)))
   else:
     trace "decodeMessages: no messages found"
   ok(msgs)
 
 proc encodeRpcMsg*(msg: RPCMsg, anonymize: bool): seq[byte] =
   trace "encodeRpcMsg: encoding message", msg = msg.shortLog()
-  var pb = initProtoBuffer()
+  var pb = initProtoBuffer(maxSize = uint.high)
   for item in msg.subscriptions:
     pb.write(1, item)
   for item in msg.messages:
@@ -324,7 +325,7 @@ proc encodeRpcMsg*(msg: RPCMsg, anonymize: bool): seq[byte] =
 
 proc decodeRpcMsg*(msg: seq[byte]): ProtoResult[RPCMsg] {.inline.} =
   trace "decodeRpcMsg: decoding message", msg = msg.shortLog()
-  var pb = initProtoBuffer(msg)
+  var pb = initProtoBuffer(msg, maxSize = uint.high)
   var rpcMsg = ok(RPCMsg())
   assign(rpcMsg.get().messages, ? pb.decodeMessages())
   assign(rpcMsg.get().subscriptions, ? pb.decodeSubscriptions())

libp2p/protocols/pubsub/timedcache.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/protocols/rendezvous.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/protocols/secure/noise.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/protocols/secure/plaintext.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/protocols/secure/secio.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/protocols/secure/secure.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/routing_record.nim

@@ -1,5 +1,5 @@
 # Nim-Libp2p
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/services/autorelayservice.nim

@@ -0,0 +1,133 @@
+# Nim-LibP2P
+# Copyright (c) 2023 Status Research & Development GmbH
+# Licensed under either of
+#  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
+#  * MIT license ([LICENSE-MIT](LICENSE-MIT))
+# at your option.
+# This file may not be copied, modified, or distributed except according to
+# those terms.
+
+when (NimMajor, NimMinor) < (1, 4):
+  {.push raises: [Defect].}
+else:
+  {.push raises: [].}
+
+import chronos, chronicles, times, tables, sequtils, options
+import ../switch,
+       ../protocols/connectivity/relay/[client, utils]
+
+logScope:
+  topics = "libp2p autorelay"
+
+type
+  OnReservationHandler = proc (addresses: seq[MultiAddress]) {.gcsafe, raises: [Defect].}
+
+  AutoRelayService* = ref object of Service
+    running: bool
+    runner: Future[void]
+    client: RelayClient
+    numRelays: int
+    relayPeers: Table[PeerId, Future[void]]
+    relayAddresses: Table[PeerId, seq[MultiAddress]]
+    backingOff: seq[PeerId]
+    peerAvailable: AsyncEvent
+    onReservation: OnReservationHandler
+    rng: ref HmacDrbgContext
+
+proc reserveAndUpdate(self: AutoRelayService, relayPid: PeerId, selfPid: PeerId) {.async.} =
+  while self.running:
+    let
+      rsvp = await self.client.reserve(relayPid).wait(chronos.seconds(5))
+      relayedAddr = rsvp.addrs.mapIt(
+        MultiAddress.init($it & "/p2p-circuit").tryGet())
+      ttl = rsvp.expire.int64 - times.now().utc.toTime.toUnix
+    if ttl <= 60:
+      # A reservation under a minute is basically useless
+      break
+    if relayPid notin self.relayAddresses or self.relayAddresses[relayPid] != relayedAddr:
+      self.relayAddresses[relayPid] = relayedAddr
+      if not self.onReservation.isNil():
+        self.onReservation(concat(toSeq(self.relayAddresses.values)))
+    await sleepAsync chronos.seconds(ttl - 30)
+
+method setup*(self: AutoRelayService, switch: Switch): Future[bool] {.async, gcsafe.} =
+  let hasBeenSetUp = await procCall Service(self).setup(switch)
+  if hasBeenSetUp:
+    proc handlePeerJoined(peerId: PeerId, event: PeerEvent) {.async.} =
+      trace "Peer Joined", peerId
+      if self.relayPeers.len < self.numRelays:
+        self.peerAvailable.fire()
+    proc handlePeerLeft(peerId: PeerId, event: PeerEvent) {.async.} =
+      trace "Peer Left", peerId
+      self.relayPeers.withValue(peerId, future):
+        future[].cancel()
+    switch.addPeerEventHandler(handlePeerJoined, Joined)
+    switch.addPeerEventHandler(handlePeerLeft, Left)
+    await self.run(switch)
+  return hasBeenSetUp
+
+proc manageBackedOff(self: AutoRelayService, pid: PeerId) {.async.} =
+  await sleepAsync(chronos.seconds(5))
+  self.backingOff.keepItIf(it != pid)
+  self.peerAvailable.fire()
+
+proc innerRun(self: AutoRelayService, switch: Switch) {.async, gcsafe.} =
+  while true:
+    # Remove relayPeers that failed
+    let peers = toSeq(self.relayPeers.keys())
+    for k in peers:
+      if self.relayPeers[k].finished():
+        self.relayPeers.del(k)
+        self.relayAddresses.del(k)
+        if not self.onReservation.isNil():
+          self.onReservation(concat(toSeq(self.relayAddresses.values)))
+        # To avoid ddosing our peers in certain conditions
+        self.backingOff.add(k)
+        asyncSpawn self.manageBackedOff(k)
+
+    # Get all connected relayPeers
+    self.peerAvailable.clear()
+    var connectedPeers = switch.connectedPeers(Direction.Out)
+    connectedPeers.keepItIf(RelayV2HopCodec in switch.peerStore[ProtoBook][it] and
+                            it notin self.relayPeers and
+                            it notin self.backingOff)
+    self.rng.shuffle(connectedPeers)
+
+    for relayPid in connectedPeers:
+      if self.relayPeers.len() >= self.numRelays:
+        break
+      self.relayPeers[relayPid] = self.reserveAndUpdate(relayPid, switch.peerInfo.peerId)
+
+    if self.relayPeers.len() > 0:
+      await one(toSeq(self.relayPeers.values())) or self.peerAvailable.wait()
+    else:
+      await self.peerAvailable.wait()
+      await sleepAsync(200.millis)
+
+method run*(self: AutoRelayService, switch: Switch) {.async, gcsafe.} =
+  if self.running:
+    trace "Autorelay is already running"
+    return
+  self.running = true
+  self.runner = self.innerRun(switch)
+
+method stop*(self: AutoRelayService, switch: Switch): Future[bool] {.async, gcsafe.} =
+  let hasBeenStopped = await procCall Service(self).stop(switch)
+  if hasBeenStopped:
+    self.running = false
+    self.runner.cancel()
+  return hasBeenStopped
+
+proc getAddresses*(self: AutoRelayService): seq[MultiAddress] =
+  result = concat(toSeq(self.relayAddresses.values))
+
+proc new*(T: typedesc[AutoRelayService],
+          numRelays: int,
+          client: RelayClient,
+          onReservation: OnReservationHandler,
+          rng: ref HmacDrbgContext): T =
+  T(numRelays: numRelays,
+    client: client,
+    onReservation: onReservation,
+    peerAvailable: newAsyncEvent(),
+    rng: rng)

libp2p/services/hpservice.nim

@@ -1,32 +0,0 @@
-when (NimMajor, NimMinor) < (1, 4):
-  {.push raises: [Defect].}
-else:
-  {.push raises: [].}
-
-import ../switch
-import chronos
-import std/tables
-
-type
-  HPService* = ref object of Service
-    newPeerHandler: PeerEventHandler
-
-proc askPeer(s: Switch, peerId: PeerId): Future[void] {.async.} =
-  echo "Asking peer " & $(peerId)
-
-proc h(switch: Switch) =
-  for p in switch.peerStore[AddressBook].book.keys:
-    discard askPeer(switch, p)
-
-method setup*(self: HPService, switch: Switch) {.async.} =
-  self.newPeerHandler = proc (peerId: PeerId, event: PeerEvent): Future[void] =
-    return askPeer(switch, peerId)
-
-  switch.connManager.addPeerEventHandler(self.newPeerHandler, PeerEventKind.Joined)
-
-method run*(self: HPService, switch: Switch) {.async, gcsafe, public.} =
-  h(switch)
-
-method stop*(self: HPService, switch: Switch) {.async, gcsafe, public.} =
-  if not isNil(self.newPeerHandler):
-    switch.connManager.removePeerEventHandler(self.newPeerHandler, PeerEventKind.Joined)

libp2p/signed_envelope.nim

@@ -1,5 +1,5 @@
 # Nim-Libp2p
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/stream/bufferstream.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/stream/chronosstream.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/stream/connection.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/stream/lpstream.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))
@@ -173,6 +173,8 @@ method closed*(s: LPStream): bool {.base, public.} =
 method atEof*(s: LPStream): bool {.base, public.} =
   s.isEof
 
+method queuedSendBytes*(s: LPStream): int {.base.} = 0
+
 method readOnce*(
   s: LPStream,
   pbytes: pointer,

libp2p/switch.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))
@@ -77,12 +77,25 @@ type
       services*: seq[Service]
 
     Service* = ref object of RootObj
+      inUse: bool
 
-method setup*(self: Service, switch: Switch) {.base, async, gcsafe, public.} = discard
 
-method run*(self: Service, switch: Switch) {.base, async, gcsafe, public.} = discard
+method setup*(self: Service, switch: Switch): Future[bool] {.base, async, gcsafe.} =
+  if self.inUse:
+    warn "service setup has already been called"
+    return false
+  self.inUse = true
+  return true
 
-method stop*(self: Service, switch: Switch) {.base, async, gcsafe, public.} = discard
+method run*(self: Service, switch: Switch) {.base, async, gcsafe.} =
+  doAssert(false, "Not implemented!")
+
+method stop*(self: Service, switch: Switch): Future[bool] {.base, async, gcsafe.} =
+  if not self.inUse:
+    warn "service is already stopped"
+    return false
+  self.inUse = false
+  return true
 
 proc addConnEventHandler*(s: Switch,
                           handler: ConnEventHandler,
@@ -117,8 +130,8 @@ method addTransport*(s: Switch, t: Transport) =
   s.transports &= t
   s.dialer.addTransport(t)
 
-method addService*(switch: Switch, service: Service) =
-  switch.services.add(service)
+proc connectedPeers*(s: Switch, dir: Direction): seq[PeerId] =
+  s.connManager.connectedPeers(dir)
 
 proc isConnected*(s: Switch, peerId: PeerId): bool {.public.} =
   ## returns true if the peer has one or more
@@ -135,17 +148,23 @@ method connect*(
   s: Switch,
   peerId: PeerId,
   addrs: seq[MultiAddress],
-  forceDial = false): Future[void] {.public.} =
+  forceDial = false,
+  reuseConnection = true): Future[void] {.public.} =
   ## Connects to a peer without opening a stream to it
 
-  s.dialer.connect(peerId, addrs, forceDial)
+  s.dialer.connect(peerId, addrs, forceDial, reuseConnection)
 
 method connect*(
   s: Switch,
-  addrs: seq[MultiAddress]): Future[PeerId] =
+  address: MultiAddress,
+  allowUnknownPeerId = false): Future[PeerId] =
   ## Connects to a peer and retrieve its PeerId
+  ##
+  ## If the P2P part is missing from the MA and `allowUnknownPeerId` is set
+  ## to true, this will discover the PeerId while connecting. This exposes
+  ## you to MiTM attacks, so it shouldn't be used without care!
 
-  s.dialer.connect(addrs)
+  s.dialer.connect(address, allowUnknownPeerId)
 
 method dial*(
   s: Switch,
@@ -278,6 +297,10 @@ proc stop*(s: Switch) {.async, public.} =
   trace "Stopping switch"
 
   s.started = false
+
+  for service in s.services:
+    discard await service.stop(s)
+
   # close and cleanup all connections
   await s.connManager.close()
 
@@ -302,7 +325,7 @@ proc stop*(s: Switch) {.async, public.} =
       a.cancel()
 
   for service in s.services:
-    await service.stop(s)
+    discard await service.stop(s)
 
   await s.ms.stop()
 
@@ -311,6 +334,10 @@ proc stop*(s: Switch) {.async, public.} =
 proc start*(s: Switch) {.async, gcsafe, public.} =
   ## Start listening on every transport
 
+  if s.started:
+    warn "Switch has already been started"
+    return
+
   trace "starting switch for peer", peerInfo = s.peerInfo
   var startFuts: seq[Future[void]]
   for t in s.transports:
@@ -342,8 +369,7 @@ proc start*(s: Switch) {.async, gcsafe, public.} =
   await s.ms.start()
 
   for service in s.services:
-    await service.setup(s)
-    await service.run(s)
+    discard await service.setup(s)
 
   s.started = true
 

libp2p/transcoder.nim

@@ -1,5 +1,5 @@
 # Nim-Libp2p
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/transports/tcptransport.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))
@@ -246,7 +246,8 @@ method accept*(self: TcpTransport): Future[Connection] {.async, gcsafe.} =
 method dial*(
   self: TcpTransport,
   hostname: string,
-  address: MultiAddress): Future[Connection] {.async, gcsafe.} =
+  address: MultiAddress,
+  peerId: Opt[PeerId] = Opt.none(PeerId)): Future[Connection] {.async, gcsafe.} =
   ## dial a peer
   ##
 

libp2p/transports/tortransport.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))
@@ -188,7 +188,8 @@ proc dialPeer(
 method dial*(
   self: TorTransport,
   hostname: string,
-  address: MultiAddress): Future[Connection] {.async, gcsafe.} =
+  address: MultiAddress,
+  peerId: Opt[PeerId] = Opt.none(PeerId)): Future[Connection] {.async, gcsafe.} =
   ## dial a peer
   ##
   if not handlesDial(address):

libp2p/transports/transport.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))
@@ -65,7 +65,8 @@ method accept*(self: Transport): Future[Connection]
 method dial*(
   self: Transport,
   hostname: string,
-  address: MultiAddress): Future[Connection] {.base, gcsafe.} =
+  address: MultiAddress,
+  peerId: Opt[PeerId] = Opt.none(PeerId)): Future[Connection] {.base, gcsafe.} =
   ## dial a peer
   ##
 
@@ -73,7 +74,8 @@ method dial*(
 
 proc dial*(
   self: Transport,
-  address: MultiAddress): Future[Connection] {.gcsafe.} =
+  address: MultiAddress,
+  peerId: Opt[PeerId] = Opt.none(PeerId)): Future[Connection] {.gcsafe.} =
   self.dial("", address)
 
 method upgradeIncoming*(

libp2p/transports/wstransport.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))
@@ -35,14 +35,18 @@ logScope:
 export transport, websock, results
 
 const
-  WsTransportTrackerName* = "libp2p.wstransport"
-
   DefaultHeadersTimeout = 3.seconds
 
 type
   WsStream = ref object of Connection
     session: WSSession
 
+method initStream*(s: WsStream) =
+  if s.objName.len == 0:
+    s.objName = "WsStream"
+
+  procCall Connection(s).initStream()
+
 proc new*(T: type WsStream,
            session: WSSession,
            dir: Direction,
@@ -129,7 +133,7 @@ method start*(
     factories = self.factories,
     rng = self.rng)
 
-  
+
   for i, ma in addrs:
     let isWss =
       if WSS.match(ma):
@@ -256,7 +260,7 @@ method accept*(self: WsTransport): Future[Connection] {.async, gcsafe.} =
 
     try:
       let
-        wstransp = await self.wsserver.handleRequest(req)
+        wstransp = await self.wsserver.handleRequest(req).wait(self.handshakeTimeout)
         isSecure = self.httpservers[index].secure
 
       return await self.connHandler(wstransp, isSecure, Direction.In)
@@ -273,6 +277,8 @@ method accept*(self: WsTransport): Future[Connection] {.async, gcsafe.} =
     debug "AsyncStream Error", exc = exc.msg
   except TransportTooManyError as exc:
     debug "Too many files opened", exc = exc.msg
+  except AsyncTimeoutError as exc:
+    debug "Timed out", exc = exc.msg
   except TransportUseClosedError as exc:
     debug "Server was closed", exc = exc.msg
     raise newTransportClosedError(exc)
@@ -286,7 +292,8 @@ method accept*(self: WsTransport): Future[Connection] {.async, gcsafe.} =
 method dial*(
   self: WsTransport,
   hostname: string,
-  address: MultiAddress): Future[Connection] {.async, gcsafe.} =
+  address: MultiAddress,
+  peerId: Opt[PeerId] = Opt.none(PeerId)): Future[Connection] {.async, gcsafe.} =
   ## dial a peer
   ##
 

libp2p/upgrademngrs/muxedupgrade.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))

libp2p/upgrademngrs/upgrade.nim

@@ -1,5 +1,5 @@
 # Nim-LibP2P
-# Copyright (c) 2022 Status Research & Development GmbH
+# Copyright (c) 2023 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))