concatAll() unexpectedly overlaps inner Observable subscriptions

[mattflix]

EDIT: Note that the following repro code always produces the expected output with RxJS 4.x. The output with RxJS 5.x is unintuitive (IMO) and can actually vary depending on the formulation of the inner observables (see further discussion of such cases in subsequent comments).

---

RxJS version: 5.5.0 (and all previous 5.x versions, seemingly)

Code to reproduce:

const { Observable } = require(`rxjs`);

function Task(name) {
  return (
    Observable
      .defer(() => {
        console.log(`begin task ${ name }`);
        return (
          Observable
            .of(`simulated work`)
            .delay(100)
            .finally(() => {
              console.log(`end task ${ name }`);
            })
          );
      })
  );
}

Observable
  .of(
    Task(`1`),
    Task(`2`),
    Task(`3`)
  )
  .concatAll()
  .subscribe();

Expected behavior:

begin task 1
end task 1
begin task 2
end task 2
begin task 3
end task 3

Actual behavior:

begin task 1
begin task 2
end task 1
begin task 3
end task 2
end task 3

Additional information:
I discovered this behavior when using concatAll() as the basis for a serialized "task queue" for executing tasks (represented as Observables) one at a time, much like my (simplified) sample code above. Some of these tasks would allocate resources in defer() (or similar manner) and release them in finally() (or similar manner).

My task queue implementation all seemed to make logical sense, yet I was running into errors caused by concurrent access to resources that were only supposed to be accessed in a mutually-exclusive manner, but somehow weren't. I was stumped, I couldn't see the the problem in my task queue or my tasks that was leading to this unexpected concurrent access.

After much head-scratching and debugging, I eventually tracked the problem to the InnerSubscriber implementation within rxjs, which contains logic that subscribes to the next sequence before unsubscribing from the previous sequence:

    InnerSubscriber.prototype._error = function (error) {
        this.parent.notifyError(error, this); // <== subscribes to the next sequence
        this.unsubscribe();
    };
    InnerSubscriber.prototype._complete = function () {
        this.parent.notifyComplete(this); // <== subscribes to the next sequence
        this.unsubscribe();
    };

Ah-ha.

This causes the "creation" logic (defer() callback or other invocation) of the next sequence to execute before the "disposal" logic (finally() or other invocation) of the previous sequence has executed, making it impossible for the inner observables managed by concatAll() to behave like serialized tasks -- since rxjs actually "overlaps" the tasks at this critical moment.

As an experiment, I simply reversed the logic:

    InnerSubscriber.prototype._error = function (error) {
        this.unsubscribe(); // <==  unsubscribe from previous sequence first!
        this.parent.notifyError(error, this);
    };
    InnerSubscriber.prototype._complete = function () {
        this.unsubscribe(); // <== unsubscribe from previous sequence first!
        this.parent.notifyComplete(this);
    };

Then, using the original "Code to reproduce" above, I actually see the expected behavior.

I don't know, however, what other ramifications this change might have on other use cases. In the case of concatAll() (and variants) however, it would definitely seem "more correct" and "less surprising" for rxjs to always unsubscribe from the previous sequence before subscribing to the next.

[mattflix]

As an aside, for the meantime, does anybody have any idea how to obtain the desired behavior (truly serialized inner observables) without having to write my own version of concatAll() or modify rxjs?

[kwonoj]

This is interesting q, while our operator does guarantee about order of emission for next, it doesn't really guarantee around how does inner subscription order is managed. I personally do not think this is something considered as a bug which shouldn't occur due to those reasons.

Put aside of debating if it's expected or not, achieving desired behavior seems somewhat tricky though. Mostly cause most operator might have similar behavior. If you can guarantee inner observables are not overlapping when emitted, you could may consider to use switch - i.e

... same Task()

Observable
  .of(
    Task(`1`),
    Task(`2`),
    Task(`3`)
  )
  .switch()
  .subscribe();

will gives you each task execution occur after teardown, cause nature of switch operator unsubscribes prior inner (which executes teardown) then move to next. But also if inner have collapsing emission and you can't lose any, switch might not be fit for you. Otherwise some complex thing like actually delay subsequent task until each of task completes. that'll work but guess will quite verbose.

[mattflix]

I agree that one might not view this behavior as a "bug" with regard to the combined emissions of the inner observables, since the emissions still occur as expected, independent of any subtleties in the subscribe/unsubscribe order of the inner observables.

That said... in cases where the only thing that matters about the inner observables is their execution (esp. if they represent tasks, or periods of time, etc. have no emissions, and only complete/error), there is strong case to be made that the operation of "concatenation" should actually take care to execute the inner observables in a truly "back to back" fashion.

So, I still find it "surprising", and quite unintuitive really, that the subscription of the next observable is overlapped with the teardown of the previous. Certainly, in my case, I wrote a bunch of code (my task queue) thinking that concatAll() couldn't possibly work that way -- as it wasn't congruent with the notion of "concatenation" in my mind.

Per my experiment, the subscribe/unsubscribe order within InnerSubscriber seems (at least at first glance) to be completely immaterial to the proper functioning of concatAll(), or any other operator, with regard to the combined emissions. The choice of order seems to have been arbitrary?

And, if that choice was indeed arbitrary, and could actually be re-made more purposefully, then choosing to explicitly avoid overlapping subscriptions within InnerSubscriber, or at least for concatAll() (and variants), could be viewed as the "least surprising", more intuitive, and potentially more useful implementation.

I hope that change can be considered.

[kwonoj]

I'll leave other chime in for opinions, but my personal opinion is still inner subscription management is internals of operator implementation and it's not expected to behave in some way. (Still, it's personal though)

Again put aside, I feel like this issue's crux is you'd like to queue certain task in given order, including it's teardown - am I understanding right? in those case, have you considered to use complete for indication of teardown for each innerobservable instead of unsubscription? For example if we reuse your examples and make rough pseudo

const { Observable } = require(`rxjs`);

function Task(name) {
  return (
    Observable
      .defer(() => {
        console.log(`begin task ${ name }`);
        return (
          Observable
            .of(`simulated work`)
            .delay(100)
            .materialize()
            .do((x) => {
              if (x.kind === 'C') {
                console.log(`end task ${ name }`);
              }
            })
            .dematerialize()
          );
      })
  );
}

Observable
  .of(
    Task(`1`),
    Task(`2`),
    Task(`3`)
  )
  .concatAll()
  .subscribe();

you'll get

begin task 1
end task 1
begin task 2
end task 2
begin task 3
end task 3

since teardown will execute on completion of each notification metadata.

[mattflix]

Thanks for the idea. Seems like it could work. One note: The inner observables may not always complete "cleanly" though, so error would also have to be "hooked" for this work.

Still, this feels quite unintuitive, and unergonomic to have to resort to such esoteric interfaces (materialize()) to successfully manage observable "lifecycle" issues, when higher-level, more unintuitive interfaces (e.g, finally()) could be used instead -- if it weren't for a seemingly arbitrary internal implementation choice within InnerSubscriber.

[kwonoj]

may not always complete "cleanly" though, so error would also have to be"hooked" for this work.

this doesn't matter, cause metadata allows you to access either error or complete, (or next value if needed) without blow up observable chain.

Still, this feels quite unintuitive, and unergonomic to have to resort...

for this I've expressed my opinion sufficiently, will leave other member's opinion around.

[mattflix]

If you can guarantee inner observables are not overlapping when emitted, you could may consider to use switch

switch() is not suitable because I don't want to "abort" the current task just because another has arrived. The purpose of the queue is to execute the current task to completion (or error) before allowing the next task to begin.

Indeed, if I could already "guarantee inner observables are not overlapping when emitted", then I would not need a queue.

[kwonoj]

switch() is not suitable

yeah, ignore that - I've already backed out those.

[kwonoj]

I vaguely recall discussed about some operator's subscription behavior in last core meeting, maybe these operator is one of the cases? /cc @benlesh

[mattflix]

An alternative solution might be to provide another operator, perhaps named serializeAll() that does guarantee subscription/unsubscription order. But, this would seem to be needlessly duplicative, as it would be concatAll() in virtually all ways but name.

[mattflix]

Also note that you do (somewhat unexpectedly ;) get the expected output when queuing up synchronous tasks.

That is, make the following small change in my original example code:

function Task(name) {
  return (
    Observable
      .defer(() => {
        console.log(`begin task ${ name }`);
        return (
          Observable
            .of(`simulated work`)
            // .delay(100) <== remove the asynchronous element
            .finally(() => {
              console.log(`end task ${ name }`);
            })
          );
      })
  );
}

...and the output is as expected.

This is another way in which the current behavior of concatAll() is surprising, and fighting the user's intuition about what "concatenation of observables" should always mean (in my opinion).

[mattflix]

Lastly, if it provides any motivation, or consideration of this behavior as a bug, note that I see the expected output from my repro code in all cases (discussed so far) when running with RxJS 4.x.

---

I have added an "EDIT" about this to the original issue description above.

[benlesh]

It's because the complete of the previous observable is what triggers subscription to the next observable in the queue. This is more of a quirk of the behavior of finally.

complete basically executes completion callback and THEN unsubscribes (which triggers finally).

To get the behavior you want, you'll need to use do or tap like type({ complete() { console.log('this is done'); } })

We could look into triggering the next subscription in concat, merge, etc after the unsubscribe of the previous observable. That seems reasonable.