fix(core): possible memory leak when using server side events

[zhengjitf]

Closes: #11601

PR Checklist

Please check if your PR fulfills the following requirements:

• The commit message follows our guidelines: https://github.com/nestjs/nest/blob/master/CONTRIBUTING.md
• Tests for the changes have been added (for bug fixes / features)
• Docs have been added / updated (for bug fixes / features)

PR Type

What kind of change does this PR introduce?

• Bugfix
• Feature
• Code style update (formatting, local variables)
• Refactoring (no functional changes, no api changes)
• Build related changes
• CI related changes
• Other... Please describe:

What is the current behavior?

Issue Number: #11601

What is the new behavior?

Does this PR introduce a breaking change?

• Yes
• No

Other information

[coveralls]

Pull Request Test Coverage Report for Build 872c4fb8-4a0a-4449-b755-a6d8ef9c9301

Details

• 4 of 4 (100.0%) changed or added relevant lines in 2 files are covered.
• No unchanged relevant lines lost coverage.
• Overall coverage increased (+0.02%) to 92.138%

---

Totals
Change from base Build e8e3c154-2e4b-4603-8729-3ff01d3ab650:	0.02%
Covered Lines:	6739
Relevant Lines:	7314

---

💛 - Coveralls

[micalevisk]

@zhengjitf can you share a minimum reproduction of this potential bug?

[zhengjitf]

@zhengjitf can you share a minimum reproduction of this potential bug?

@micalevisk Sorry for not having much description for this PR, a minimum is mentioned from #11601 (comment), and I committed a unit test to reproduce this issue before fixing it, (it might be not intuitive?).

I’d like to describe my changes, along with investigation:

The MaxListenersExceededWarning comes out when the line 120 code runs too frequently.

nest/packages/core/router/sse-stream.ts

Lines 119 to 121 in 87cf25a

	if (!this.write(message, 'utf-8', cb)) {
	this.once('drain', cb);
	} else {

Then I noticed there is a limitation intention in the following code, but it doesn't work, which I understand is to limit only emitting the value when meanwhile the previous writing is completed (the previous drain event is emitted), for this intention it should be replaced with exhaustMap, but I think concatMap is more reasonable, in a serialized fashion waiting for the previous writing completed, as a result, will only one listener binds drain event simultaneously.

nest/packages/core/router/router-response-controller.ts

Lines 131 to 136 in 87cf25a

	debounce(
	message =>
	new Promise<void>(resolve =>
	stream.writeMessage(message, () => resolve()),
	),
	),

Based on my understanding, this issue will happen when trying to write a message but meanwhile the previous writing is not completed, this leads to binding more listeners to the drain event until exceeds the max listener limitation (default as 10).

A realistic case is mentioned in #11601:

I use Server-Sent Events for one of my routes and then push db changes for subscribed users.

For this case, the issue is likely to happen when the db changes frequently and each pushed event data is large.

PS: I didn't run into this issue, I'm trying to implement a SSE route with POST method and did some investigation in the issues panel and source code, then found this issue and hope to make some efforts to address it.

I would appreciate it if any misunderstanding or mistakes from me are pointed out.

[benjGam]

It seems to be a clever alternative to do not overload the process.

[kamilmysliwiec]

lgtm