Possible EventEmitter memory leak detected

[leobel]

Hello, I've an http server using the same InteractionContext to handle multiple request creating instances of TxSubmissionClient (without closing any of this clients after response), after several requests I end up with this warning:

(node: 8) MaxListenersExceededWarning: Possible EventEmitter memory leak detected. 
11 message listeners added to [WebSocket]. Use emitter.setMaxListeners() to increase limit

@KtorZ already mention the use of submitTx method (without creating any client) and sharing the InteractionContext between requests, but taking a look to the source code of TxSubmissionClient it looks like a wrapper of the underlying websocket with a couple of convenient methods such as: shutdown and internal ensureSocketIsOpen:

/**
 * Create a client for submitting signed transactions to underlying Cardano chain.
 *
 * @category Constructor
 **/
export const createTxSubmissionClient = async (
  context: InteractionContext
): Promise<TxSubmissionClient> => {
  const { socket } = context
  return Promise.resolve({
    context,
    evaluateTx: (bytes) => {
      ensureSocketIsOpen(socket)
      return evaluateTx(context, bytes)
    },
    submitTx: (bytes) => {
      ensureSocketIsOpen(socket)
      return submitTx(context, bytes)
    },
    shutdown: () => new Promise(resolve => {
      ensureSocketIsOpen(socket)
      socket.once('close', resolve)
      socket.close()
    })
  } as TxSubmissionClient)
}

So, I'm hitting the limit for a legitimate reason in this case and just need to increase (the default value is 10) the maximum listener value, or there is something else underneath we should care about?

[KtorZ]

(cc @rhyslbw if you have any bandwidth to look into this, otherwise I'll have a look when I am back from vacations 😶)

[leobel]

here is my code:

import { Injectable } from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import { createInteractionContext, createTxSubmissionClient, InteractionContext, getServerHealth, createConnectionObject, Connection, ConnectionConfig } from '@cardano-ogmios/client';
import { TxSubmissionClient, submitTx } from '@cardano-ogmios/client/dist/TxSubmission';

@Injectable()
export class OgmiosService {
    connection: Connection;
    connectionConfig: ConnectionConfig;
    context: InteractionContext;
    constructor(private configService: ConfigService) {
        const port = this.configService.get<number>('PORT') || 1337;
        const host = this.configService.get<string>('HOST') || 'localhost';
        const tls = this.configService.get<string>('TLS') == 'true';
        this.connectionConfig = { host, port, tls };
        this.connection = createConnectionObject(this.connectionConfig);
    }

    async connect(): Promise<void> {
        this.context = await createInteractionContext(
            err => {
                this.context = null;
                console.error('Connection error', err);
            },
            (code: number, reason: string) => {
                this.context = null;
                console.log(`Connection closed. Code: ${code}, Reason: ${reason}`);
            },
            { connection: this.connectionConfig, interactionType: 'LongRunning' }
        );

    }


    async isConnected(): Promise<boolean> {
        return !!this.context && ((await getServerHealth({ connection: this.connection })) as any).connectionStatus == "connected";
    }

    async submitTx(cbor: string): Promise<TxSubmissionClient> {
        // await submitTx(this.context, cbor);
        const client = await createTxSubmissionClient(this.context);
        return client;
    }
}

Here I'm handling each request:

const isConnected = await this.ogmiosService.isConnected();
if (!isConnected) {
    console.log('Connecting to ws ...');
    await this.ogmiosService.connect();
    console.log('Connected to ws :)')
}
await this.ogmiosService.submitTx(cbor);
console.log('Tx Success');         

[KtorZ]

I've been looking into but I am unable to reproduce (using your code above, firing 100s of requests); though I've only tried with failing transactions; The WebSocket listeners are properly registered and de-registered as far as I can tell. There's something I found thought but which should only cause an issue when the response gotten from the server is unexpected. If you're submitting well-formed transactions (valid or invalid) that shouldn't impact you.

• So I am a bit puzzled. What version of Node.js are you on?
• And, what's the rate at which you're submitting transactions?

One thing I can think of is that, maybe there are "too many" in-flight transactions; the listener currently only gets de-registered after a transaction has been submitted; it's quite fast but it still take some time. So possibly, if you're sending many transactions fast, it fills up the slots for the listeners.

[leobel]

Hi @KtorZ

What version of Node.js are you on?

I'm using node v14.17.6 on NestJS server v8.0.0

what's the rate at which you're submitting transactions?

I'm firing 500 requests, it doesn't mean the 500 are hitting the server at the same time since they are enqueued/dequeued at batches of at most 10 or waiting 1 sec whatever comes first (the throughput is dictated by the blockchain).

[javiertc]

Hi @KtorZ could you help us with this? with 100 requests you won't get the error, you'll have to make 500. Thanks

[KtorZ]

@javiertc, when I say "100s" I mean, hundreds of transactions. I think I've been up to 1000 but most requests were resolved promptly so I may never had more than 10 listeners at once.

The thing is, the default limit is pretty low and, if you're indeed firing a lot of requests through the same context, then in all likelihood, the warning is benign and caused by the heavy load set by the client.

In the end, the server processes transaction sequentially, so there's no much point parallelizing anything on the client. So I see two possibles paths of resolutions:

(1) increase the default limit to, say, 100 and give ways for callers to configure it (higher or lower).

(2) rework the client to not setup one listener per request, but to simply use a queue, processing requests one by one.

Option (2) gives potentially a way to apply backpressure back to the client. I mean, if the client submits way faster than the server can ingest, there's no point to keep accepting transactions. The client could throw an exception when the queue is full.

[KtorZ]

@javiertc @leobel #206 should solve your issue.

Note that this is only addressed in the TxSubmissionClient (which is now a bit more complicated than the standalone submitTx / evaluateTx functions). So, you're encouraged to instantiate a single TxSubmissionClient and use it concurrently to handle many requests. It should set up at most 2 listeners (one for SubmitTx, and one for EvaluateTx).

Please also keep in mind that behind the scene, the server still handles transaction one by one regardless and, it'll simply enqueue requests it receives while already submitting one. This means that you can possibly create a resources exhaustion if your application is sending transactions faster than the node can digest them! You're thereby very much invited to monitor either your server resources (i.e. heap) or the mempool to and apply back-pressure to your client in case things grow out of hands.

If you are able to test the branch and let me know, that'd be awesome. I've already done a bunch of testing (and was finally able to reproduce by adding some artificial delays in the request processing on the server's side) and as far as I (and tests) can tell, it works fine.

[leobel]

Hi @KtorZ thanks for sharing it! We'll take a look to the branch and let you know our results.

[KtorZ]

Closing this issue proactively with the assumption that #206 does indeed fix it. Feel free to re-open if it's not the case.