Proposal: Changes to retryWhen

[benlesh]

These would be breaking changes, but here goes:

Problem

Often with retryWhen there is a desire to reset some "state" of your retries when the stream your retrying starts receiving values successfully again. For example, in a network retry scenario where you want to exponentially step back from the first failure. Once you get a value over that network observable, you want to reset your exponential step back to the first level again, because things are going well now.

Current Solution

The current solution involves adding some peice of mutable state and altering it via side-effects:

let exponent = 2;
let result = source.retryWhen(errs => errs.switchMap(() => Observable.timer(Math.pow(2, exponent++))))
      .do(() => exponent = 0)

Proposed Change (and solution)

I propose a change that says the first successful value to "next" through the retryWhen subscriber causes the retry nofication subscription to unsubscribe, which will "reset" the state of any observable inside of there.

let result = source.retryWhen(errs => errors.switchMap((_, i) => Observable.timer(Math.pow(2, i)));

The behavior:

1. Subscribe to result
2. Subscribes to source and passes through successful values and completions
3. On error from source, the error is pumped into a Subject, and the notifier function passed to retryWhen is called with an Observable created from that Subject.
4. The observable returned from the notifier function is subscribed to.
5. On next from the notifier, the source subscription is retried.
   a. On first successful message from source, unsubscribe from notifier, then goto 3.
6. On error from the notifier, an error is sent down the chain to the result.
7. On complete from the notifier, the completion is sent down the chain to the result.

[staltz]

I took a look at retryWhen.ts, thought about your suggestions, and I think it's a reasonable breaking change.

[benlesh]

I just spoke with @abersnaze here at Netflix, and I think there's a way to do this without a breaking change.

Basically, the idea is to add an optional second parameter that fires for each next from source, and is called with the value from the next, and the subscription from the inner errors subscription.

source.retryWhen(errs => errs.switchMap((_, i) => Observable.timer(2**i)), (value, sub) => sub.unsubscribe());

Another option would be a second argument that was the same idea, but you just returned a boolean of whether or not to end the subscription.

EDIT: To clarify

A boolean variant instead, where true is "I'd like to reset my inner subscription"

source.retryWhen(errs => errs.switchMap((_, i) => Observable.timer(2**i)), (value) => true);

An async variant of the same, where a next signals the recycling of the inner subscription:

source.retryWhen(errs => errs.switchMap((_, i) => Observable.timer(2**i)), (value) => Observable.of('whatever'));

Or another, where you get a stream of all nexted values, and return a stream of notifications to retry the retry notifier (probably too confusing):

source.retryWhen(errs => errs.switchMap((_, i) => Observable.timer(2**i)), (values) => values.switchMap(() => timer(100)));

[benlesh]

Also, @stealthcode (from Netflix) had some thoughts about a different design (as a new operator). His concern with tearing down the inner subscription and starting it back up was based on the principle that that's an additional teardown and setup that you might not need. He wanted to try to keep the inner subscription .. so it would look like this:

retryWhenWithNext(notifierFactory: (errorsOrNexts: Observable<any>) => Observable<any>)

The idea here is that each call to next or error would come through the observable passed to the notifierFactory.

given:  source.retryWhenWithNext((errorsOrNexts) => errorsOrNexts.filter(x => x instanceof Error))
source ---a---b---c---d---#
errorsOrNexts ---a---b---c---d---#---a---b---c---d---#

Given that type-checking in JavaScript isn't a thing, I'm not sure how this approach would work for us, so perhaps in our case it would be better to have a stream of Notification in there for that design:

repeatWhenNotified(notifierFactory: (notifications: Observable<Notification<T>>) => Observable<any>)

It's another design, and it's interesting, because it would actually allow for retries on async actions after errors, completions or nexted values.

[benlesh]

@mattpodwysocki I'd really like your thoughts here, there's a lot in this thread.

[benlesh]

I'm of the mind, that even if we implement a repeatWhenNotified type operator, we might still want to add the optional argument to retryWhen, as it wouldn't break the API, and it would be useful. Also retryWhen has a concise use case, where repeatWhenNotified would not. On the other hand, if those notifications were restricted to errors and nexts, then it would again, and we could call it retryWhenNotified I guess.

[benlesh]

I've edited my comment above to clarify the options presented there.

[danturu]

+1

[benlesh]

I think this issue still has merit... leaving it for now.

[jiayihu]

@benlesh I'm having exactly this issue while implementing a polling library based on RxJS. Is there any current, even dirty and extremely hacky, way to workaround this issue?

EDIT: I think I've solved it using a counter to keep track of retryWhen state on last recover. The snippet, in its essentials, is the following:

let allErrorsCount = 0;
let lastRecoverCount = 0;

Observable.interval(1000)
  .switchMap(() => request$)
  .retryWhen(errors$ => {
    return errors$.scan((errorCount, err) => errorCount + 1, 0).switchMap(errorCount => {
      allErrorsCount = errorCount;
      const consecutiveErrorsCount = allErrorsCount - lastRecoverCount;
      const esponentialDelay = Math.pow(2, consecutiveErrorsCount) * 1000;

      return Observable.timer(esponentialDelay, null, scheduler);
    });
  })
  .do(() => {
    // Update the counter after success recover
    lastRecoverCount = allErrorsCount;
  });

What do you think?

---

About the proposal, if I can give an opinion, I'd prefer this option:

An async variant of the same, where a next signals the recycling of the inner subscription

That's because you can:

1. Use some Observable like Observable.timer to delay reset if you have any particular need
2. It's simple for common usage, because Observable.of(true) is enough
3. It feel consistent with current behaviour of retryWhen, where the retry is driven by an Observable

repeatWhenNotified(notifierFactory: (notifications: Observable<Notification>) => Observable)

This operator also would be very nice IMO and would respect the 3 points I've mentioned, but since it's a new operator there should be other use cases for it

[benlesh]

This is pretty old, and there hasn't been much appetite to change it. However, we did at a "reset" to the retry operator, so perhaps we might go that route at a later time. Closing for now.