fix(subscriber): don't unsubscribe self #4106

cartant · 2018-09-06T22:06:22Z

Description:

This PR fixes a problem that was introduced in #3963 and was partially fixed in #4072.

The problem was introduced with this change - to Subscriber - in #3963 where a parent subscription is unsubscribed upon a complete notification:

complete(): void {
  if (!this.isStopped) {
    this.isStopped = true;
    this._complete();
+   this._unsubscribeParentSubscription();
  }
}

The parent subscription is added to the subscriber in Observable.subscribe:

const sink = toSubscriber(observerOrNext, error, complete);

if (operator) {
  operator.call(sink, this.source);
} else {
  sink._addParentTeardownLogic(
    this.source || (config.useDeprecatedSynchronousErrorHandling && !sink.syncErrorThrowable) ?
    this._subscribe(sink) :
    this._trySubscribe(sink)
  );
}

In numerous situations, the subscription passed to _addParentTeardownLogic will be the same subscriber. That is, it will be sink.

Adding sink to itself, as a parent, introduces a bug, as sink will be unsubscribed from within the complete method - which differs from the pre-#3963 mechanism.

What follows below is a description of the PR's original implementation. This implementation was changed substantially as the PR was discussed. For an explanation of the PR's final, merged changes, skip to my comment at the bottom of this conversation.

#4072 introduced a check to ensure that a specific subscriber isn't added to itself as a parent. However, it's still possible for a subscriber that is a child of the specific subscriber to be added as a parent. And doing so will see the specific subscriber's unsubscribe called within its complete method - which can effect unwanted behaviour. See the failing test that's included in this PR.

This PR solves the problem by adding a _keepAliveCount that's incremented before unsubscribing the parent subscription and is decremented afterwards. In Subscription.unsubscribe, if _keepAliveCount is greater than zero, the subscription is not unsubscribed.

Related issue (if exists): #4095

coveralls · 2018-09-06T22:25:11Z

Coverage decreased (-0.003%) to 96.968% when pulling b817f53 on cartant:issue-4095 into 6feec15 on ReactiveX:master.

benlesh · 2018-09-13T22:53:55Z

Thank you for bringing this issue to my attention. Taking a step back, I think that _addParentTeardownLogic is probably a smell, and we likely just wanted to this.add(trustedSubscriber) on this line

benlesh

See my comment, and refer to our offline discussion.

cartant · 2018-09-14T07:37:49Z

@benlesh This PR is now a WIP, as I've made some changes to make things simpler, but the changes need to be applied to many of the operator-specific subscribers and I've only changed MergeMapSubscriber, so far.

I've removed all of the parent-teardown business and the _keepAliveCount.

The _addParentTeardownLogic function is still there - as a noop - as its used in isTrustedSubscriber. I'm not sure what you'll want to do with _addParentTeardownLogic if it's no longer needed.

Anyway, the important parts are in MergeMapSubscriber's _innerSub function, where the inner subscription is added to the destination (which should always be a trusted subscriber). And in its _complete function, where this.unsubscribe() is called.

To me this is simpler than dealing with children, parents and destinations. In the original implementation - that preceded the PR that sought to address the not-unsubscribing-from-source problem - a subscription was added to its destination so that unsubscribing the destination unsubscribes the subscription. That behaviour is now restored. And because the subscription's destination is not added to the subscription, calling this.unsubscribe() in _complete does not unsubscribe the destination - or any inner observables that were added to it.

If you think this approach is a better way to go, I can apply similar changes to the other subscribers. (It's not just the flattening subscribers. E.g. the subscribers for delay and delayWhen will need to be changed, too.)

We would need to decide the best way of adding subscriptions to the destination, as the type assertions are rather ugly:

const destination = this.destination as any as Subscription;

benlesh

Yes. This is precisely what I was suggesting when we discussed this yesterday.

benlesh · 2018-09-14T16:53:41Z

spec/operators/mergeMap-spec.ts

 import { hot, cold, expectObservable, expectSubscriptions } from '../helpers/marble-testing';

 declare const type: Function;
 declare const asDiagram: Function;

 /** @test {mergeMap} */
-describe('mergeMap', () => {
+describe.only('mergeMap', () => {


I know this is a WIP, but don't forget to remove this only.

benlesh · 2018-09-14T16:56:30Z

src/internal/operators/mergeMap.ts

@@ -140,7 +140,8 @@ export class MergeMapSubscriber<T, R> extends OuterSubscriber<T, R> {

  private _innerSub(ish: ObservableInput<R>, value: T, index: number): void {
    const innerSubscriber = new InnerSubscriber(this, undefined, undefined);
-    this.add(innerSubscriber);
+    const destination = this.destination as any as Subscription;


destination is a Subscriber, which extends Subscription... I'm not sure why any casting is necessary here... 🤔

Nope, it's a PartialObserver:

protected destination: PartialObserver<any>;

Changing destination to be:

protected destination: PartialObserver<any> | Subscriber<any>;

means the type assertion becomes:

const destination = this.destination as Subscription;

benlesh · 2018-09-14T17:00:23Z

src/internal/Subscriber.ts

@@ -158,16 +156,7 @@ export class Subscriber<T> extends Subscription implements Observer<T> {

  /** @deprecated This is an internal implementation detail, do not use. */
  _addParentTeardownLogic(parentTeardownLogic: TeardownLogic) {


We can probably just remove _addParentTeardownLogic entirely.

It's prefixed with _ which is a universal signal for "do not use".

It's been marked as "deprecated do not use" since the moment it was added.

My concern is that the name has been used in the isTrustedSubscriber test and that's been released. I wasn't entirely sure of what the affect of it's removal in a new release would be. I suppose it means a trusted subscriber could end up being re-wrapped in a using-6.3.3-with-6.3.2 scenario.

When unsubscribing a subscriber's parent, make sure that the subscriber itself is not unsubscribed. Closes ReactiveX#4095

cartant · 2018-09-14T22:08:59Z

@benlesh This should be done, now.

It's been rebased. The commits made subsequent to your comments are those that follow refactor(mergeMap): simplify.

benlesh · 2018-09-19T22:31:53Z

@cartant I think we should schedule a time to walk through this one.

Remove the mapTo and the concat to reduce the number of subscribers to make the test easier to reason with.

cartant · 2018-10-06T21:57:25Z

What follows are notes on the history of changes made in this PR.

#3963 (which was a refactoring of #2457) made changes to the way in which subscribers were wired up. The changes were made so that teardown logic for sources was called as soon as sources completed or errored.

Its changes were:

Instead of just adding a subscriber to its destination, it was added using a new method: _addParentTeardownLogic.
In Observable.prototype.subscribe, the result from _trySubscribe was also added using _addParentTeardownLogic.
When complete or error notifications occurred, subscribers would unsubscribe the teardown login added via _addParentTeardownLogic.

The major problem with this approach was that a subscriber could be added to itself, as _trySubscribe can return the subscriber that's passed to it.

This PR added a test that included this code:

const wrapped = new Observable<number>(subscriber => {
    const subscription = timer(0, asapScheduler).subscribe(subscriber);
    return () => subscription.unsubscribe();
});
wrapped.pipe(
  mergeMap(() => timer(0, asapScheduler))
).subscribe({
  next(value) { results.push(value); },
  complete() { results.push('done'); }
});

When run with the changes in #3963, the sequence of subscribers/subscriptions was:

construct Subscription@1
construct Subscriber@2
construct SafeSubscriber@3
construct MergeMapSubscriber@4
add MergeMapSubscriber@4 to Subscriber@2
construct AsapAction@5
add AsapAction@1 to MergeMapSubscriber@4
construct Subscription@6
add Subscription@6 to MergeMapSubscriber@4
next MergeMapSubscriber@4
construct InnerSubscriber@7
add InnerSubscriber@7 to MergeMapSubscriber@4 <--
construct AsapAction@8
add AsapAction@2 to InnerSubscriber@7
complete MergeMapSubscriber@4
unsubscribe Subscription@6
unsubscribe MergeMapSubscriber@4
unsubscribe AsapAction@1
unsubscribe InnerSubscriber@7
unsubscribe AsapAction@2

Note that the InnerSubscriber was added to the MergeMapSubscriber. That means that the InnerSubscriber was unsubscribed when the MergeMapSubscriber was unsubscribed. And that was happen when the source - the wrapped observable - completed.

This PR reverted the changes in #3963 and made the following changes to a number of operators:

Inner subscribers are now added to the destination subscriber.
Outer subscribers now unsubscribe upon complete and error, with their inner subscribers remaining subscribed due to their having been added to the destination.

With these changes, the test now passes and the sequence of subscribers/subscriptions is:

construct Subscription@1
construct Subscriber@2
construct SafeSubscriber@3
construct MergeMapSubscriber@4
add MergeMapSubscriber@4 to Subscriber@2
construct AsapAction@5
add AsapAction@1 to MergeMapSubscriber@4
construct Subscription@6
add Subscription@6 to MergeMapSubscriber@4
next MergeMapSubscriber@4
construct InnerSubscriber@7
add InnerSubscriber@7 to Subscriber@2 <--
construct AsapAction@8
add AsapAction@2 to InnerSubscriber@7
complete MergeMapSubscriber@4
unsubscribe MergeMapSubscriber@4
unsubscribe AsapAction@1
unsubscribe Subscription@6
next InnerSubscriber@7
next Subscriber@2
complete InnerSubscriber@7
complete Subscriber@2
unsubscribe SafeSubscriber@3
unsubscribe Subscriber@2
unsubscribe InnerSubscriber@7
unsubscribe AsapAction@2

Note that the InnerSubscriber is now added to the destination Subscriber and is therefore not unsubscribed upon completion of the source. And is able to next its value, etc.

The changes in this PR affected a number of behaviour tests in observeOn-spec.ts and in switch-spec.ts because adding the inner subscibers to the destination changed the subscription counts that those tests were checking.

benlesh requested changes Sep 13, 2018

View reviewed changes

cartant added the PR: WIP label Sep 14, 2018

benlesh requested changes Sep 14, 2018

View reviewed changes

benlesh reviewed Sep 14, 2018

View reviewed changes

benlesh mentioned this pull request Sep 14, 2018

sink._addParentTeardownLogic is not a function rxjs node 6.3.2 and 6.2.2 #4135

Closed

cartant added 19 commits September 15, 2018 08:07

test(mergeMap): add failing test

d83264d

fix(subscriber): don't unsubscribe self

83414b3

When unsubscribing a subscriber's parent, make sure that the subscriber itself is not unsubscribed. Closes ReactiveX#4095

refactor(mergeMap): simplify

80e52ee

chore(typings): use union type for destination

bc75ee2

chore(test): remove only

55e60d6

chore(test): use pipe

c2395e6

test(Subscriber): fake add too

1d02c73

test(internals): update for subscription changes

d383d24

refactor(zip): add to destination

b12c14c

refactor(delay): add to destination

0e33a56

refactor(delayWhen): add to destination

456baa9

refactor(exhaustMap): add to destination

c721cd4

refactor(expand): add to destination

21b73b0

refactor(mergeScan): add to destination

26d070b

refactor(observeOn): add to destination

dad60bf

refactor(onErrorResumeNext): add to destination

e226a77

refactor(sequenceEqual): add to destination

0044084

refactor(switchMap): add to destination

9918ab2

chore(Subscriber): remove _addParentTeardownLogic

4eb3f44

cartant force-pushed the issue-4095 branch from 1cc1070 to 4eb3f44 Compare September 14, 2018 22:08

cartant removed the PR: WIP label Sep 14, 2018

chore(test): simplify mergeMap test

b817f53

Remove the mapTo and the concat to reduce the number of subscribers to make the test easier to reason with.

benlesh approved these changes Sep 25, 2018

View reviewed changes

benlesh merged commit e623ec6 into ReactiveX:master Sep 25, 2018

lock bot locked as resolved and limited conversation to collaborators Nov 6, 2018

cartant deleted the issue-4095 branch September 24, 2020 07:08

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

fix(subscriber): don't unsubscribe self #4106

fix(subscriber): don't unsubscribe self #4106

cartant commented Sep 6, 2018 •

edited

coveralls commented Sep 6, 2018 •

edited

benlesh commented Sep 13, 2018

benlesh left a comment

cartant commented Sep 14, 2018 •

edited

benlesh left a comment

benlesh Sep 14, 2018

benlesh Sep 14, 2018

cartant Sep 14, 2018

cartant Sep 14, 2018

benlesh Sep 14, 2018

cartant Sep 14, 2018

cartant commented Sep 14, 2018 •

edited

benlesh commented Sep 19, 2018

cartant commented Oct 6, 2018 •

edited

		@@ -158,16 +156,7 @@ export class Subscriber<T> extends Subscription implements Observer<T> {

		/** @deprecated This is an internal implementation detail, do not use. */
		_addParentTeardownLogic(parentTeardownLogic: TeardownLogic) {

fix(subscriber): don't unsubscribe self #4106

fix(subscriber): don't unsubscribe self #4106

Conversation

cartant commented Sep 6, 2018 • edited

coveralls commented Sep 6, 2018 • edited

benlesh commented Sep 13, 2018

benlesh left a comment

Choose a reason for hiding this comment

cartant commented Sep 14, 2018 • edited

benlesh left a comment

Choose a reason for hiding this comment

benlesh Sep 14, 2018

Choose a reason for hiding this comment

benlesh Sep 14, 2018

Choose a reason for hiding this comment

cartant Sep 14, 2018

Choose a reason for hiding this comment

cartant Sep 14, 2018

Choose a reason for hiding this comment

benlesh Sep 14, 2018

Choose a reason for hiding this comment

cartant Sep 14, 2018

Choose a reason for hiding this comment

cartant commented Sep 14, 2018 • edited

benlesh commented Sep 19, 2018

cartant commented Oct 6, 2018 • edited

cartant commented Sep 6, 2018 •

edited

coveralls commented Sep 6, 2018 •

edited

cartant commented Sep 14, 2018 •

edited

cartant commented Sep 14, 2018 •

edited

cartant commented Oct 6, 2018 •

edited