Provide an async do on retry callback

[remk]

Motivation

The Retry.doOnRetry allow to register a synchronous callback. At the time being if we want to use an async operation in this callback which finish before the next retry is attempted we have to block on it. This can cause some locks and was not usable. This was reported in the gitter channel reactor/reactor by @mbaechler.

Hi there, I have a problem with using retryWhen on my Mono. Basically, what I want to do is:

    put a REST resource somewhere
    If failing with a "parent resource doesn't exist", retry using Retry.onlyIf and doOnRetry creating the parent resource

However, doOnRetry is called synchronously, so the REST request I perform here requires a call to block, triggering in some cases a starvation of my threads

Do you know any way to implement that in a different way?

Desired solution

So it would be good to be able to register an async callback with an equivalent to Retry.doOnRetry for example :

 public Retry<T> onRetryWithMono(Function<? super RetryContext<T>, Mono<?>> onRetryMono) 

[rouazana]

Here is a similar issue: https://stackoverflow.com/questions/55924849/on-error-do-another-call-and-retry-in-webflux/55937290

This would avoid the ugly workaround shown is answers.

[simonbasle]

At first this felt like bending the intent of Retry builder, which was initially thought around the combination of counter + delays. doOnRetry could also be seen as a side-effect, that are not supposed to introduce blocking behavior / latency.

That said, why not having this way of further customizing the retry attempt. Two caveats:

1. we need to come up with a better name for it
2. the documentation must be very explicit that this changes the delay between two attempts. eg. if you use that + linearBackoff(Duration.ofSeconds(1)), the backoff is not guaranteed to be 1s anymore (depending on how long the custom Mono takes to terminate).

In balancing whether or not to add it, let's also consider that:

1. the "ugly" code in the stackoverflow question stems from a more complex retry case than the gitter one, and can probably be made a bit more palatable by putting the Function in a constant, or as a class method reference
2. Simpler cases where only an action + single retry is needed, the onErrorResume approach below can be used:

operation.retry(e -> isTransient(e), 10)
    .onErrorResume(PreconditionException.class, e -> prepareOperation().then(operation));

[simonbasle]

The two alternatives with pure core operators:

public final Function<Flux<Throwable>, Publisher<String>> retryOrFixPrecondition(int max, Predicate<Throwable> canRetry,
		Predicate<Throwable> isPreconditionError, Mono<String> fixPrecondition) {
	return ft -> ft
			.index()
			.flatMap(idxError -> {
				if (isPreconditionError.test(idxError.getT2())) {
					return fixPrecondition;
				}
				if (!canRetry.test(idxError.getT2()) || idxError.getT1() > max) {
					throw Exceptions.propagate(idxError.getT2());
				}
				return Mono.just("retry");
			});
}

@Test
public void retrySpecialErrorWithPrecondition() {
	//simulate a source with transient errors + a precondition error that needs
	//specific handling then retry
	AtomicInteger count = new AtomicInteger(0);
	AtomicBoolean prerequisite = new AtomicBoolean();
	Mono<String> operation = Mono.defer(() -> {
		int c = count.getAndIncrement();
		if (c < 2) {
			throw new IllegalStateException("TRANSIENT FAILURE " + c);
		}
		if (prerequisite.get()) {
			return Mono.just("SUCCESS");
		}
		throw new IllegalArgumentException("PREREQUISITE");
	});

	//simulate a long running asynchronous operation that updates state to match
	//the precondition
	Mono<String> prepare = Mono.delay(Duration.ofSeconds(3))
	                           .doOnNext(ignored -> prerequisite.set(true))
	                           .thenReturn("retry");

	//predicate for errors that can be retried
	Predicate<Throwable> shouldRetry = p -> p instanceof IllegalStateException
			&& p.getMessage() != null
			&& p.getMessage().startsWith("TRANSIENT");

	//predicate for the error that reports precondition is not met
	Predicate<Throwable> isMissingPrerequisite = p -> p instanceof IllegalArgumentException
			&& "PREREQUISITE".equals(p.getMessage());

	//APPROACH 1, capable of retrying further in case of transient failures after
	//the prerequisite state is updated
	Mono<String> retryWhen = operation
			.retryWhen(retryOrFixPrecondition(10, shouldRetry, isMissingPrerequisite, prepare));

	//APPROACH 2, if you want to retry other exceptions but update prerequisite state
	// and re-attempt the operation only once
	Mono<String> retryResume = operation
			.retry(10, shouldRetry)
			.onErrorResume(isMissingPrerequisite, e -> prepare.then(operation));


	//testing all this
	retryWhen.as(StepVerifier::create)
	         .expectNext("SUCCESS")
	         .verifyComplete();
	count.set(0); prerequisite.set(false);
	retryResume.as(StepVerifier::create)
	           .expectNext("SUCCESS")
	           .verifyComplete();
}

[remk]

Thank you for your suggestions, but it means we should re-implement the backoff and jitter we rely on with Retry.

[simonbasle]

@remk I see. We'll discuss this. Also note that we're thinking about bringing the builder pattern into the core library in parallel, so maybe that new feature should be part of this effort core-side rather than addons-side.