Unsubscribe from a combineLatest doesn't unsubscribe from underlying observables

[RoqueIT]

Describe the bug

Since the version 7.x.x, the behavior seems to have changes in the unsubscription mecanism for combineLatest: unsubscribe from a combineLatest doesn't unsubscribe from underlying observables

Expected behavior

Unsubscribe from a combineLatest observable should unsubscribe from underlying observables

Reproduction code

import { describe, expect, jest, test } from '@jest/globals';
import { combineLatest, Observable, Subscription } from 'rxjs';

test('Must unsubscribe everywhere', () => {
  let observable1$ = new Observable<string>();
  let subscription1 = new Subscription();
  const observable1Spy = jest
    .spyOn(observable1$, 'subscribe')
    .mockReturnValue(subscription1);
  const unsubscription1Spy = jest.spyOn(subscription1, 'unsubscribe');

  let observable2$ = new Observable<string>();
  let subscription2 = new Subscription();
  const observable2Spy = jest
    .spyOn(observable2$, 'subscribe')
    .mockReturnValue(subscription2);
  const unsubscription2Spy = jest.spyOn(subscription1, 'unsubscribe');

  expect(observable1Spy).toHaveBeenCalledTimes(0);
  expect(unsubscription1Spy).toHaveBeenCalledTimes(0);
  expect(subscription1.closed).toBeFalsy();
  expect(observable2Spy).toHaveBeenCalledTimes(0);
  expect(unsubscription2Spy).toHaveBeenCalledTimes(0);
  expect(subscription2.closed).toBeFalsy();

  let combined$ = combineLatest([observable1$, observable2$]);
  const subscription3 = combined$.subscribe((value) => {});

  expect(subscription3.closed).toBeFalsy();
  expect(observable1Spy).toHaveBeenCalledTimes(1);
  expect(unsubscription1Spy).toHaveBeenCalledTimes(0);
  expect(subscription1.closed).toBeFalsy();
  expect(observable2Spy).toHaveBeenCalledTimes(1);
  expect(unsubscription2Spy).toHaveBeenCalledTimes(0);
  expect(subscription2.closed).toBeFalsy();

  subscription3.unsubscribe();

  expect(subscription3.closed).toBeTruthy();
  expect(observable1Spy).toHaveBeenCalledTimes(1);
  expect(unsubscription1Spy).toHaveBeenCalledTimes(1);
  expect(subscription1.closed).toBeTruthy();
  expect(observable2Spy).toHaveBeenCalledTimes(1);
  expect(unsubscription2Spy).toHaveBeenCalledTimes(1);
  expect(subscription2.closed).toBeTruthy();
});

Reproduction URL

https://stackblitz.com/edit/rxjs-issue-7318?file=test%2Funsubscribe.spec.ts&terminal=test

Version

7.x.x

Environment

No response

Additional context

Change RXJS to version "^6.6.7" in package.json, and the test will pass

We can see in the code that subscription to an underlying observable is not kept in memory anymore:

rxjs/src/internal/observable/combineLatest.ts

Line 214 in 2947583

source.subscribe(

[voliva]

I think this is because of the test setup.... you are mocking the subscribe functions of your observable so that it returns a subscription, but then it's not calling the actual .subscribe() of the observable.

Note that observables do get unsubscribed when using regular operators, you can observe this behavior in the example provided here: https://codesandbox.io/s/rxjs-playground-3rkq8z?file=/src/index.ts

When using an observable through an operator, it doesn't receive a user-land observer through the subscribe method. Instead, it receives an internal Subscriber object that manages the observable's lifecycle. It's what the operate function in the next line of the one you point out does:

rxjs/src/internal/observable/combineLatest.ts

Lines 214 to 217 in 2947583

	source.subscribe(
	operate({
	destination,
	next: (value) => {

So, to clarify, internally, the subscription returned by subscribe is not utilized. Instead, a different mechanism is employed to unsubscribe.

This issue highlights another instance where Subscriber, an implementation detail, leaks out. This relates to a suggestion I was involved in, namely #7250. In my opinion, adopting that suggestion would have resulted in a more preferable and less leaky API for observables. However, this suggestion faced significant resistance from @benlesh .

Interestingly, one of the arguments against that suggestion was that RxJS' operators wouldn't work with "standard observables."
Ironically, this scenario reveals that RxJS doesn't even invoke the .unsubscribe() method on the subscription returned by .subscribe(). So any other Observable library that is unaware of the internal workings of Subscriber would also encounter compatibility issues with the existing operators. Which I think it's ok, RxJS operators should work with RxJS observables, and then you can have functions to interop between different Observable types, but I find it funny that one of the main reasons for rejecting that suggestion is something that applies to the current state of RxJS' operators.

I think it would really help if you can show a use case where this is problematic without mocking out the .subscribe() function of the observable, which is messing up with the internals.

[josepot]