Possible EventTarget memory leak detected.

[noahfigueras]

• Version:
  "libp2p": "^0.45.9"

• Platform:
  Linux nf 6.2.0-26-generic Update libp2p-swarm to version 0.23.0 🚀 #26~22.04.1-Ubuntu SMP PREEMPT_DYNAMIC Thu Jul 13 16:27:29 UTC 2 x86_64 x86_64 x86_64 GNU/Linux

• Subsystem:
  Dialer

Severity:

Medium

Description:

I have a setup with multiple nodes connected to relay. Usually on start works with no problems after 1 day or so of runtime. We usually experience this warning an all of the nodes and we see a spike of 100% cpu constant usage after that, which slows down dramatically other running processes. (I'm not sure those are related, but everything seems to happen after that warning log).
Also, we experienced that nodes lose pubsub messages after that as well.

It's the Dialer Relay that logs that warning.

Error:

(node:2880541) MaxListenersExceededWarning: Possible EventTarget memory leak detected. 11 relay:removed listeners added to ReservationStore. Use ev ents.setMaxListeners() to increase limit

Steps to reproduce the error:

Relay Setup:

async function main () {
  const node = await createLibp2p({
    peerId: await createFromJSON(rid),
    addresses: {
      listen: ['/ip4/127.0.0.1/tcp/5040/ws'],
      announce: ['/dns4/relay.smoothly.money/tcp/443/wss/']
    },
    transports: [
      webSockets(),
    ],
    connectionEncryption: [
      noise()
    ],
    streamMuxers: [
      yamux({
        maxMessageSize: 1 << 20 // 1 MB
      })
    ],
    services: {
      identify: identifyService(),
      relay: circuitRelayServer({
        reservations: {
          defaultDataLimit: BigInt(1 << 20), // the default maximum number of bytes that can be transferred over a relayed connection
        }
      }),
      pubsub: gossipsub({ allowPublishToZeroPeers: true })
    },
    peerDiscovery: [
      pubsubPeerDiscovery({
        interval: 1000
      })
    ]
  })

  console.log(`Node started with id ${node.peerId.toString()}`)
  console.log('Listening on:')
  node.getMultiaddrs().forEach((ma) => console.log(ma.toString()))
  node.addEventListener('peer:discovery', async (evt) => {
        const { id } = evt.detail;
        console.log("Found:", id);
        await node.dial(id);
  })
}

main()

Dialer Setup:

  async createNode(): Promise<void> {
    try {
      const node = await createLibp2p({
        transports: [
          webSockets(),
          circuitRelayTransport({
            discoverRelays: 1
          })
        ],
        streamMuxers: [
          yamux({
            maxMessageSize: 1 << 20
          })
        ],
        connectionEncryption: [
          noise()
        ],
        services: {
          pubsub: gossipsub({ allowPublishToZeroPeers: true }),
          identify: identifyService()
        },
        peerDiscovery: [
          bootstrap({
            list: this.bootstrapers
          }),
          pubsubPeerDiscovery({
            interval: 1000,
          })
        ],
      });

      // Personal_id channel
      node.services.pubsub.subscribe(`${node.peerId.toString()}`)
      // Checkpoint check
      node.services.pubsub.subscribe('checkpoint')

      // Track peers
      node.addEventListener('peer:discovery', async (evt) => {
        const { addresses, id } = evt.detail as any;

        // Avoids relay nodes
        if(addresses.length > 0) {
          const peer: Peer = {
            address: addresses[0].multiaddr,
            id: id
          };

          // Exists peer?
          if(this._findPeer(peer.id) === null) {
            this.peers.push(peer);
            console.log("Discovered Peer:", id.toString(), "total:", this.peers.length);
          }
        }
      })

      // Handle pubsub messages
      node.services.pubsub.addEventListener('message', (evt) => {
        const { from } = evt.detail as any;
        if(evt.detail.topic === `${node.peerId.toString()}`) {
          const data = Buffer.from(evt.detail.data).toString();
          if(data === 'sync') {
            const peer = this._findPeer(from);
            if(peer) {
              this.dialPeerSync(peer.address);
            }
          }
        } else if(evt.detail.topic === 'checkpoint'){
          const { root, epoch } = JSON.parse(uint8ArrayToString(evt.detail.data));
          console.log('checkpoint:', from, root, epoch);
          this.consensus.addVote(from, root, epoch);
        }
      })

      // Handle stream muxing from peer:sync 
      node.handle('/sync:peer', async ({ stream }) => {
        let db: DB = this.db;
        await pipe(
          stream,
          async function (source) {
            // Add data
            for await (const msg of source) {
              let validator = JSON.parse(uint8ArrayToString(msg.subarray()));
              await db.insert(validator.index, validator);
            }
          }
        )
        stream.close();
      })

      await node.start();
      this.node = node;
    } catch(err: any) {
      console.log(err);
    }
  }