Error: stream reset when pinging /p2p-circuit relay addresses concurrently

[JoshuaCWebDeveloper]

Version:
0.41.0

Platform:
Ubuntu:
Linux ********* 5.15.79.1-microsoft-standard-WSL2 #1 SMP Wed Nov 23 01:01:46 UTC 2022 x86_64 x86_64 x86_64 GNU/Linux

Chrome:
108.0.5359.125 (Official Build) (64-bit) (cohort: Stable)

Subsystem:
Unsure, likely a combination of ping and circuit relay

Severity:

Medium High

Description:

• What you did
  I attempted to concurrently ping a list of multiaddrs that included /p2p-circuit addresses.

• What happened
  Many of the ping attempts failed with the error Error: stream reset (sometimes with different stream errors)

• What you expected to happen
  All of the ping attempts should have succeeded, since these were all healthy addresses.

Steps to reproduce the error:

Find a reproduction with instructions at: https://github.com/JoshuaCWebDeveloper/libp2p-concurrent-ping

Key points:

• The list of addresses being pinged includes a /p2p-circuit address (this error does not happen if none of the addresses are relay addresses).
• The list of addresses are pinged concurrently (this error does not happen if many addresses are pinged sequentially).

[bakhshandeh]

Same problem when more than 30 parallel streams are open between 2 nodes.

Here is the error:

Error: stream reset
    at Object.reset (node_modules/@libp2p/mplex/dist/src/stream.js:105:33)
    at MplexStreamMuxer._handleIncoming (node_modules/@libp2p/mplex/dist/src/mplex.js:253:24)
    at MplexStreamMuxer.sink (node_modules/@libp2p/mplex/dist/src/mplex.js:156:36)
    at processTicksAndRejections (node:internal/process/task_queues:96:5)
  
code: 'ERR_STREAM_RESET'

Sample code to reproduce the issue:

import { tcp } from "@libp2p/tcp";
import { mplex } from "@libp2p/mplex";
import { noise } from "@chainsafe/libp2p-noise";
import { createLibp2p } from "libp2p";
import { pipe } from "it-pipe";
import { toString as uint8ArrayToString } from "uint8arrays/to-string";
import { fromString as uint8ArrayFromString } from "uint8arrays/from-string";

const createNode = async () => {
  const node = await createLibp2p({
    addresses: {
      listen: ["/ip4/127.0.0.1/tcp/0"],
    },
    transports: [tcp()],
    streamMuxers: [mplex()],
    connectionEncryption: [noise()],
  });

  await node.start();
  return node;
};

(async () => {
  const listenerNode = await createNode();

  // Log a message when we receive a connection
  listenerNode.connectionManager.addEventListener("peer:connect", (evt) => {
    const connection = evt.detail;
    console.log("Received dial from:", connection.remotePeer.toString());
  });

  // echo protocol
  await listenerNode.handle("/echo/1.0.0", async ({ stream }) => {
    await pipe(stream.source, stream.sink);
    stream.close();
  });

  // ================= DIALER ============

  const dialerNode = await createNode();
  async function callProcess() {
    for (let i = 0; i < 50; i++) {
      console.log(`i: ${i}`);
      // Dial the listener node
      const stream = await dialerNode.dialProtocol(
        listenerNode.getMultiaddrs()[0],
        "/echo/1.0.0"
      );
      await pipe(
        // Source data
        [uint8ArrayFromString(`hey [${i}]`)],
        // Write to the stream, and pass its output to the next function
        stream,
        // Sink function
        async function(source) {
          // For each chunk of data
          for await (const data of source) {
            // Output the data
            console.log("Received echo:", uint8ArrayToString(data.subarray()));
          }
        }
      );
      // stream.close();
    }
  }

  // parallel streams
  const PARALLEL_CALLS = 40; // 30 works fine
  await Promise.all(new Array(PARALLEL_CALLS).fill(0).map((i) => callProcess()));
  console.log("all done");
})();

"dependencies": {
    "@chainsafe/libp2p-gossipsub": "^5.3.0",
    "@chainsafe/libp2p-noise": "^10.0.1",
    "@libp2p/bootstrap": "^5.0.0",
    "@libp2p/kad-dht": "^6.0.0",
    "@libp2p/mdns": "^5.1.1",
    "@libp2p/mplex": "^7.1.1",
    "@libp2p/pubsub-peer-discovery": "^7.0.0",
    "@libp2p/tcp": "^6.0.0",
    "@libp2p/websockets": "^5.0.2",
    "@nodeutils/defaults-deep": "^1.1.0",
    "it-concat": "^2.0.0",
    "it-drain": "^2.0.0",
    "it-last": "^2.0.0",
    "it-length-prefixed": "^3.1.0",
    "it-pair": "^1.0.0",
    "it-pipe": "^1.1.0",
    "it-pushable": "^1.4.0",
    "libp2p": "^0.41.0",
    "multiformats": "^9.6.4"
  }```

[bakhshandeh]

Same error when using js-libp2p-yamux as the streamMuxer.
So the issue could be on tcp module not mplex.

[achingbrain]

You are hitting stream limits. When you register your handler you need to increase the default limits:

// echo protocol
  await listenerNode.handle("/echo/1.0.0", async ({ stream }) => {
    await pipe(stream.source, stream.sink);
    stream.close();
  }, {
  maxInboundStreams: /* something sensible */,
  maxOutboundStreams: /* something sensible */
});

The value of something sensible needs to be tuned for your application.

See LIMITS.md#stream-limits for more.

[github-actions]

Oops, seems like we needed more information for this issue, please comment with more details or this issue will be closed in 7 days.

[github-actions]

This issue was closed because it is missing author input.