Stop using aborted event for KibanaRequest.events.aborted$

[pgayvallet]

Summary

Fix #125240

Fix a problem causing KibanaRequest.events.aborted$ to not emit in scenarios where it should, when used within endpoints consuming the payload (basically most of our POST or PUT endpoints).

This was caused by a regression deliberate (and undocumented) change in the way node's IncomingMessage internally works. See nodejs/node#40775 for more context.

Also has the upside to stop using that aborted event which is flagged as deprecated since node v16.12.0 (https://nodejs.org/docs/latest-v16.x/api/http.html#event-abort)

Checklist

• Unit or functional tests were updated or added to match the most common scenarios

[pgayvallet]

This is the workaround recommended way to distinguish if a response was properly completed or aborted when a close event is fired, given nodejs/node#40775 (comment)

[pgayvallet]

I initially tried to use only one source fromEvent observable, two destination subjects, and manually emitting to these depending on isCompleted, but the implementation was, in the end, more complex, and more vulnerable to edge cases such as errors. Given we were already using fromEvent twice before these changes, I think it's acceptable anyway.

[afharo]

RxJS noob question here: do we still get any benefit from having one fromEvent observable?

Suggested change

	const finished$ = fromEvent<void>(request.raw.res, 'close').pipe(
	filter(() => {
	return isCompleted(request);
	}),
	first()
	);
	// the response's underlying connection was terminated prematurely
	const aborted$ = fromEvent<void>(request.raw.res, 'close').pipe(
	filter(() => {
	return !isCompleted(request);
	}),
	first(),
	takeUntil(finished$)
	);
	const closed$ = fromEvent<void>(request.raw.res, 'close');
	const finished$ = closed$.pipe(
	filter(() => {
	return isCompleted(request);
	}),
	first()
	);
	// the response's underlying connection was terminated prematurely
	const aborted$ = closed$.pipe(
	filter(() => {
	return !isCompleted(request);
	}),
	first(),
	takeUntil(finished$)
	);

I'm not sure about the internal implementation of RxJS, but it looks like it'd create only one listener to the close event? Question is... would it work as expected?

[gsoldevila]

From the fromEvent documentation:

Every time resulting Observable is subscribed, event handler function will be registered to event target on given event type. When that event fires, value passed as a first argument to registered function will be emitted by output Observable. When Observable is unsubscribed, function will be unregistered from event target.

To answer @afharo 's question, I created a small StackBlitz to confirm that 2 listeners are created / attached anyway with the proposed approach.

If you really want to have a single listener to the original event, there is no need to create Subjects and emit to them. I believe you can simply use the share operator:

const closed$ = fromEvent<void>(request.raw.res, 'close').pipe(share());

[gsoldevila]

I wonder whether the finished$ and aborted$ Observables should also have the replay behavior.

Your proposed finished$ does not have it anymore (as opposed to finish$), which means if somebody subscribes to the event after it has happened they will just miss it and might get stuck.

IMHO there's no harm in replaying the last event (if it exists) for this type of network related events (we do have a first(), which ensures we'll only be getting one).

[pgayvallet]

Your proposed finished$ does not have it anymore (as opposed to finish$),

That's true, but finished$ is only internal and the public completed$ observable subscribes to it 'instantly' (no return to the ev loop, so no risk for the event to fire in the middle), and has a replay effect, so I think it was unnecessary to have it in two places.

Regarding aborted$, there was no replay effect before the PR, and adding one did break tests, so I felt it was safer to KISS and preserve the original behavior as much as possible.

[gsoldevila]

Also, perhaps we can reorder the operators and use first() first, this way there is no need to use takeUntil().

Suggested change

	const finished$ = fromEvent<void>(request.raw.res, 'close').pipe(
	filter(() => {
	return isCompleted(request);
	}),
	first()
	);
	// the response's underlying connection was terminated prematurely
	const aborted$ = fromEvent<void>(request.raw.res, 'close').pipe(
	filter(() => {
	return !isCompleted(request);
	}),
	first(),
	takeUntil(finished$)
	);
	const closed$ = fromEvent<void>(request.raw.res, 'close').pipe(
	shareReplay(1),
	first(),
	);
	const finished$ = $closed.pipe(
	filter(() => {
	return isCompleted(request);
	})
	);
	// the response's underlying connection was terminated prematurely
	const aborted$ = $closed.pipe(
	filter(() => {
	return !isCompleted(request);
	})
	);

[pgayvallet]

nevermind regarding adding replay to aborted$, it was probably very late yesterday. changes in 19b120b and df3173b

[pgayvallet]

Also, perhaps we can reorder the operators

Definitely better, yea. PR updated.

[elasticmachine]

Pinging @elastic/kibana-core (Team:Core)

[afharo]

LGTM! Just have a nit question (also RxJS noob question): https://github.com/elastic/kibana/pull/126184/files#r812712848

[kibana-ci]

💚 Build Succeeded

• Buildkite Build
• Commit: be0a2f6
• Storybooks Preview
• Webpack Bundle Analyzer

Metrics [docs]

✅ unchanged

History

• 💚 Build #25643 succeeded 302f5aa

To update your PR or re-run it, just comment with:
@elasticmachine merge upstream

[kibanamachine]

💚 All backports created successfully

Status	Branch	Result
✅	8.1
✅	8.0

Note: Successful backport PRs will be merged automatically after passing CI.

Questions ?

Please refer to the Backport tool documentation