chore: investigate if possible to send many messages in a short amount of time (throughput is very limited in js-waku)

[x48115]

Problem

When attempting to send many messages back to back we run into an error: ERR_TOO_MANY_OUTBOUND_PROTOCOL_STREAMS

Steps to reproduce

import {
  createLightNode,
  waitForRemotePeer,
  createEncoder,
  utf8ToBytes,
} from "@waku/sdk";
import { Protocols } from "@waku/interfaces";
const contentTopic = "/light-guide/1/message/proto";
const encoder = createEncoder({ contentTopic });
const createNode = async () => {
  const node = await createLightNode({
    defaultBootstrap: true,
  });
  await node.start();
  await waitForRemotePeer(node, [Protocols.LightPush, Protocols.Filter]);
  setInterval(() => {
    node.lightPush.send(encoder, utf8ToBytes("ping"));
  }, 300);
};

createNode();

Expected Results

Expected to be able to send many messages back to back with no error (I can do this with libp2p directly with both floodsub and gossipsub implementations

Actual results

file:///Users/user/src/waku/node_modules/libp2p/dist/src/upgrader.js:325
                        const err = errCode(new Error(`Too many outbound protocol streams for protocol "${protocol}" - limit ${outgoingLimit}`), codes.ERR_TOO_MANY_OUTBOUND_PROTOCOL_STREAMS);
                                            ^

Error: Too many outbound protocol streams for protocol "/vac/waku/lightpush/2.0.0-beta1" - limit 64
    at ConnectionImpl.newStream [as _newStream] (file:///Users/user/src/waku/node_modules/libp2p/dist/src/upgrader.js:325:45)
    at process.processTicksAndRejections (node:internal/process/task_queues:95:5)
    at async ConnectionImpl.newStream (file:///Users/user/src/waku/node_modules/libp2p/dist/src/connection/index.js:55:24)
    at async LightPush.send (file:///Users/user/src/waku/node_modules/@waku/core/dist/lib/light_push/index.js:29:24) {
  code: 'ERR_TOO_MANY_OUTBOUND_PROTOCOL_STREAMS'
}

Node.js v20.1.0

Libp2p example with no issues

/* eslint-disable no-console */

import { noise } from "@chainsafe/libp2p-noise";
import { yamux } from "@chainsafe/libp2p-yamux";
import { gossipsub } from "@chainsafe/libp2p-gossipsub";
import { mplex } from "@libp2p/mplex";
import { tcp } from "@libp2p/tcp";
import { createLibp2p } from "libp2p";
import { identifyService } from "libp2p/identify";
import { fromString as uint8ArrayFromString } from "uint8arrays/from-string";
import { toString as uint8ArrayToString } from "uint8arrays/to-string";

const createNode = async () => {
  const node = await createLibp2p({
    addresses: {
      listen: ["/ip4/0.0.0.0/tcp/0"],
    },
    transports: [tcp()],
    streamMuxers: [yamux(), mplex()],
    connectionEncryption: [noise()],
    services: {
      pubsub: gossipsub(),
      identify: identifyService(),
    },
  });
  return node;
};

(async () => {
  const topic = "news";
  const [node1, node2] = await Promise.all([createNode(), createNode()]);
  await node1.peerStore.patch(node2.peerId, {
    multiaddrs: node2.getMultiaddrs(),
  });
  await node1.dial(node2.peerId);

  node1.services.pubsub.subscribe(topic);
  node1.services.pubsub.addEventListener("message", (evt) => {
    console.log(
      `node1 received: ${uint8ArrayToString(evt.detail.data)} on topic ${
        evt.detail.topic
      }`
    );
  });

  setInterval(() => {
    node2.services.pubsub
      .publish(topic, uint8ArrayFromString("Bird bird bird, bird is the word!"))
      .catch((err) => {
        console.error(err);
      });
  }, 10);
})();

Note

This is mentioned in a couple places:
ChainSafe/js-libp2p-gossipsub#306
ChainSafe/js-libp2p-gossipsub#293

Suggested fix

Currently lightPush.send makes a new stream for every method. In the comments in the two issues above it seems js-libp2p's expectation is that a protocol should only have one open stream and send all messages over that stream instead of opening a new stream per message.

js-waku/packages/core/src/lib/light_push/index.ts

Line 49 in 89392db

const stream = await this.newStream(peer);

[fryorcraken]

This is interesting as I have in the past tried to re-used streams but failed to do so. I think I mainly experimented with store and not light push

[danisharora099]

while it shouldn't matter, but with js-libp2p example gossipsub is directly being used (no wait for ack), while lightpush uses req-res model (per stream) and requires to wait for ack.

if OP wants to parallelise and send multiple messages using lightpush over js-libp2p, configuring this options to increase max outbound streams should help: https://github.com/libp2p/js-libp2p/blob/5ffa7a74d5b972bdac387782d6010b6a19558600/packages/pubsub-gossipsub/src/stream.ts#L8 might help

we can't reuse streams with lightpush because each request is sent on one stream, and libp2p will close stream after a response is receieved.

@x48115 pls let us know if that helps. happy to keep this open until satisfaction is obtained.

[fryorcraken]

To add on that, I have tried in the past to re-use the same stream on req-resp protocols such as store but libp2p closes the stream once the response is received. Maybe it's something that changed.

Also note that if you send several messages on the outbound stream with light push, you will expect several responses (for the ack). The whole point of being able to create several stream is this exact multiplexing need AFAIK.

As mentioned above, you should be able to pass libp2p options to createLightNode to increase the number of streams (maxOutboundStreams).

[weboko]

Moving to Triage to discuss any potential work which might include graceful handling of sending multiple requests at once.
Potentially can be perceived as part of #2154

[weboko]

Let's check:

• what capabilities js-waku has - how many messages per second etc;
• how RLN limit interferes with amount of messages being send (should we throw if it is higher than RLN limit);
• if we can parallelize sending many lightPush;

Additional action point: if increase to outbound streams influence it anyhow. This issue does not improve reliability.