Repository: reactor/reactor-core
Issue: Mono and Flux with onErrorContinue hangs on network errors (thrown by Netty)

When I use an `onErrorContinue` on a `Mono` or a `Flux` that gets its input from a `WebClient`, the whole chain hangs in case a network error is thrown. I've seen this behaviour with errors caused by a connection timeout, read timeout, unknown host and a refused connection.

Please note that I created this bug in reactor-core, because this is the place where `onErrorContinue` is defined. I suspect that this bug is not specific to the netty http client, but I haven't tested that. 

## Expected Behavior
I expect the program to continue and not to cause any thread to hang.

## Actual Behavior
The thread that executes the Mono (or Flux) blocks forever.

## Steps to Reproduce

You can use the following junit 5 tests (one for Mono and one for Flux) to reproduce this issue. This test instantiates a WebClient with a 200ms response timeout. This client is then used to do a call to "https://httpstat.us/200?sleep=500" that will return a 200 response only after 500ms. This way we force a response timeout to be thrown.

A failing test means that the `block()` did not return any result within 5 seconds.

```java
public class NettyReactorTest {
    private final WebClient webClient;

    public NettyReactorTest() {
        final ReactorClientHttpConnector httpConnector = new ReactorClientHttpConnector(
                HttpClient.create().responseTimeout(Duration.ofMillis(200)));

        this.webClient = WebClient.builder()
                .baseUrl("https://httpstat.us")
                .clientConnector(httpConnector)
                .build();
    }

    @Test
    public void testMonoOnErrorContinueDoesNotHangOnTimeouts() {
        Mono<ResponseEntity<Void>> call = webClient.get()
                .uri("/200?sleep=500")
                .retrieve()
                .toBodilessEntity()
                .onErrorContinue((e, o) -> {
                });

        Assertions.assertTimeoutPreemptively(Duration.ofSeconds(5), () -> call.block());
    }

    @Test
    public void testFluxOnErrorContinueDoesNotHangOnTimeouts() {
        Mono<ResponseEntity<Void>> call = webClient.get()
                .uri("/200?sleep=500")
                .retrieve()
                .toBodilessEntity();

        Mono<List<ResponseEntity<Void>>> calls = Flux.fromIterable(List.of(1, 2, 3))
                .flatMap(i -> call)
                .onErrorContinue((e, o) -> {})
                .collectList();

        Assertions.assertTimeoutPreemptively(Duration.ofSeconds(5), () -> calls.block());
    }
}
```

## Extra information

When we take a thread dump a little while after we run the test (that blocks) we see that the thread that calls `block()` is stuck on the `Mono.block()` operator:
```
junit-timeout-thread-1
  jdk.internal.misc.Unsafe.park(boolean, long) Unsafe.java (native)
  java.util.concurrent.locks.LockSupport.park(Object) LockSupport.java:194
  java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt() AbstractQueuedSynchronizer.java:885
  java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(int) AbstractQueuedSynchronizer.java:1039
  java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(int) AbstractQueuedSynchronizer.java:1345
  java.util.concurrent.CountDownLatch.await() CountDownLatch.java:232
  reactor.core.publisher.BlockingSingleSubscriber.blockingGet() BlockingSingleSubscriber.java:87
  reactor.core.publisher.Mono.block() Mono.java:1707
  NettyReactorTest.lambda$testMonoOnErrorContinueDoesNotHangOnTimeouts$1(Mono) NettyReactorTest.java:52
  NettyReactorTest$$Lambda$529.get()
  org.junit.jupiter.api.AssertTimeout.lambda$assertTimeoutPreemptively$4(AtomicReference, ThrowingSupplier) AssertTimeout.java:138
  org.junit.jupiter.api.AssertTimeout$$Lambda$530.call()
  java.util.concurrent.FutureTask.run() FutureTask.java:264
  java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor$Worker) ThreadPoolExecutor.java:1128
  java.util.concurrent.ThreadPoolExecutor$Worker.run() ThreadPoolExecutor.java:628
  java.lang.Thread.run() Thread.java:829
```

## Possible Solution

Not a solution, but more of a way to mitigate the impact.

I am aware that `onErrorContinue` is a specialised operator and that for my specific case I could (and should) have used `onErrorResume`.  Indeed, replacing `onErrorContinue` with `onErrorResume` in the above tests makes the problem go away. So, perhaps I am just another case of someone using it wrongly and "tripping up on it" (https://github.com/reactor/reactor-core/issues/2184).

Yet, this seemingly innocent code causes a rather nasty issue that should either be:
* fixed if this is indeed a bug or 
* prevented if this is an unjust use of the `onErrorContinue` operator. 

For the latter, I like @simonbasle's suggestion to move this operator to an `unsafe()` subgroup. This would have prevented me from using it and use `onErrorResume` instead as suggested in the documentation.

## Your Environment
* Reactor version(s) used: 3.4.22
* Other relevant libraries versions (eg. `netty`, ...):
** netty: 1.0.22
** spring / spring-web: 5.3.22
* JVM version (`java -version`): OpenJDK Runtime Environment Zulu11.58+15-CA (build 11.0.16+8-LTS)
* OS and version (eg `uname -a`): Darwin Kernel Version 21.6.0: Mon Aug 22 20:17:10 PDT 2022; root:xnu-8020.140.49~2/RELEASE_X86_64 x86_64 (macOS Monterey)


Changed files:
reactor-core/src/main/java/reactor/util/retry/RetrySpec.java, reactor-core/src/test/java/reactor/core/publisher/FluxRetryWhenTest.java, reactor-core/src/main/java/reactor/util/retry/RetryBackoffSpec.java

diff --git a/reactor-core/src/main/java/reactor/util/retry/RetryBackoffSpec.java b/reactor-core/src/main/java/reactor/util/retry/RetryBackoffSpec.java
index 3283bd781..540501b5c 100644
--- a/reactor-core/src/main/java/reactor/util/retry/RetryBackoffSpec.java
+++ b/reactor-core/src/main/java/reactor/util/retry/RetryBackoffSpec.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2020-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2020-2022 VMware Inc. or its affiliates, All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -31,6 +31,7 @@ import reactor.core.publisher.Mono;
 import reactor.core.scheduler.Scheduler;
 import reactor.core.scheduler.Schedulers;
 import reactor.util.annotation.Nullable;
+import reactor.util.context.Context;
 import reactor.util.context.ContextView;
 
 /**
@@ -539,69 +540,73 @@ public final class RetryBackoffSpec extends Retry {
 	@Override
 	public Flux<Long> generateCompanion(Flux<RetrySignal> t) {
 		validateArguments();
-		return t.concatMap(retryWhenState -> {
-			//capture the state immediately
-			RetrySignal copy = retryWhenState.copy();
-			Throwable currentFailure = copy.failure();
-			long iteration = isTransientErrors ? copy.totalRetriesInARow() : copy.totalRetries();
-
-			if (currentFailure == null) {
-				return Mono.error(new IllegalStateException("Retry.RetrySignal#failure() not expected to be null"));
-			}
-
-			if (!errorFilter.test(currentFailure)) {
-				return Mono.error(currentFailure);
-			}
-
-			if (iteration >= maxAttempts) {
-				return Mono.error(retryExhaustedGenerator.apply(this, copy));
-			}
-
-			Duration nextBackoff;
-			try {
-				nextBackoff = minBackoff.multipliedBy((long) Math.pow(2, iteration));
-				if (nextBackoff.compareTo(maxBackoff) > 0) {
+		return Flux.deferContextual(cv ->
+		   t.contextWrite(cv)
+			.concatMap(retryWhenState -> {
+				//capture the state immediately
+				RetrySignal copy = retryWhenState.copy();
+				Throwable currentFailure = copy.failure();
+				long iteration = isTransientErrors ? copy.totalRetriesInARow() : copy.totalRetries();
+
+				if (currentFailure == null) {
+					return Mono.error(new IllegalStateException("Retry.RetrySignal#failure() not expected to be null"));
+				}
+
+				if (!errorFilter.test(currentFailure)) {
+					return Mono.error(currentFailure);
+				}
+
+				if (iteration >= maxAttempts) {
+					return Mono.error(retryExhaustedGenerator.apply(this, copy));
+				}
+
+				Duration nextBackoff;
+				try {
+					nextBackoff = minBackoff.multipliedBy((long) Math.pow(2, iteration));
+					if (nextBackoff.compareTo(maxBackoff) > 0) {
+						nextBackoff = maxBackoff;
+					}
+				}
+				catch (ArithmeticException overflow) {
 					nextBackoff = maxBackoff;
 				}
-			}
-			catch (ArithmeticException overflow) {
-				nextBackoff = maxBackoff;
-			}
-
-			//short-circuit delay == 0 case
-			if (nextBackoff.isZero()) {
-				return RetrySpec.applyHooks(copy, Mono.just(iteration),
+
+				//short-circuit delay == 0 case
+				if (nextBackoff.isZero()) {
+					return RetrySpec.applyHooks(copy, Mono.just(iteration),
+							syncPreRetry, syncPostRetry, asyncPreRetry, asyncPostRetry);
+				}
+
+				ThreadLocalRandom random = ThreadLocalRandom.current();
+
+				long jitterOffset;
+				try {
+					jitterOffset = nextBackoff.multipliedBy((long) (100 * jitterFactor))
+							.dividedBy(100)
+							.toMillis();
+				}
+				catch (ArithmeticException ae) {
+					jitterOffset = Math.round(Long.MAX_VALUE * jitterFactor);
+				}
+				long lowBound = Math.max(minBackoff.minus(nextBackoff)
+						.toMillis(), -jitterOffset);
+				long highBound = Math.min(maxBackoff.minus(nextBackoff)
+						.toMillis(), jitterOffset);
+
+				long jitter;
+				if (highBound == lowBound) {
+					if (highBound == 0) jitter = 0;
+					else jitter = random.nextLong(highBound);
+				}
+				else {
+					jitter = random.nextLong(lowBound, highBound);
+				}
+				Duration effectiveBackoff = nextBackoff.plusMillis(jitter);
+				return RetrySpec.applyHooks(copy, Mono.delay(effectiveBackoff,
+								backoffSchedulerSupplier.get()),
 						syncPreRetry, syncPostRetry, asyncPreRetry, asyncPostRetry);
-			}
-
-			ThreadLocalRandom random = ThreadLocalRandom.current();
-
-			long jitterOffset;
-			try {
-				jitterOffset = nextBackoff.multipliedBy((long) (100 * jitterFactor))
-				                          .dividedBy(100)
-				                          .toMillis();
-			}
-			catch (ArithmeticException ae) {
-				jitterOffset = Math.round(Long.MAX_VALUE * jitterFactor);
-			}
-			long lowBound = Math.max(minBackoff.minus(nextBackoff)
-			                                     .toMillis(), -jitterOffset);
-			long highBound = Math.min(maxBackoff.minus(nextBackoff)
-			                                    .toMillis(), jitterOffset);
-
-			long jitter;
-			if (highBound == lowBound) {
-				if (highBound == 0) jitter = 0;
-				else jitter = random.nextLong(highBound);
-			}
-			else {
-				jitter = random.nextLong(lowBound, highBound);
-			}
-			Duration effectiveBackoff = nextBackoff.plusMillis(jitter);
-			return RetrySpec.applyHooks(copy, Mono.delay(effectiveBackoff,
-					backoffSchedulerSupplier.get()),
-					syncPreRetry, syncPostRetry, asyncPreRetry, asyncPostRetry);
-		});
+			})
+		    .contextWrite(c -> Context.empty())
+		);
 	}
 }
diff --git a/reactor-core/src/main/java/reactor/util/retry/RetrySpec.java b/reactor-core/src/main/java/reactor/util/retry/RetrySpec.java
index 55c3ae989..710ff4de8 100644
--- a/reactor-core/src/main/java/reactor/util/retry/RetrySpec.java
+++ b/reactor-core/src/main/java/reactor/util/retry/RetrySpec.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2020-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2020-2022 VMware Inc. or its affiliates, All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -26,6 +26,7 @@ import java.util.function.Predicate;
 import reactor.core.Exceptions;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.util.context.Context;
 import reactor.util.context.ContextView;
 
 /**
@@ -353,25 +354,30 @@ public final class RetrySpec extends Retry {
 
 	@Override
 	public Flux<Long> generateCompanion(Flux<RetrySignal> flux) {
-		return flux.concatMap(retryWhenState -> {
-			//capture the state immediately
-			RetrySignal copy = retryWhenState.copy();
-			Throwable currentFailure = copy.failure();
-			long iteration = isTransientErrors ? copy.totalRetriesInARow() : copy.totalRetries();
+		return Flux.deferContextual(cv ->
+			flux
+					.contextWrite(cv)
+					.concatMap(retryWhenState -> {
+						//capture the state immediately
+						RetrySignal copy = retryWhenState.copy();
+						Throwable currentFailure = copy.failure();
+						long iteration = isTransientErrors ? copy.totalRetriesInARow() : copy.totalRetries();
 
-			if (currentFailure == null) {
-				return Mono.error(new IllegalStateException("RetryWhenState#failure() not expected to be null"));
-			}
-			else if (!errorFilter.test(currentFailure)) {
-				return Mono.error(currentFailure);
-			}
-			else if (iteration >= maxAttempts) {
-				return Mono.error(retryExhaustedGenerator.apply(this, copy));
-			}
-			else {
-				return applyHooks(copy, Mono.just(iteration), doPreRetry, doPostRetry, asyncPreRetry, asyncPostRetry);
-			}
-		});
+						if (currentFailure == null) {
+							return Mono.error(new IllegalStateException("RetryWhenState#failure() not expected to be null"));
+						}
+						else if (!errorFilter.test(currentFailure)) {
+							return Mono.error(currentFailure);
+						}
+						else if (iteration >= maxAttempts) {
+							return Mono.error(retryExhaustedGenerator.apply(this, copy));
+						}
+						else {
+							return applyHooks(copy, Mono.just(iteration), doPreRetry, doPostRetry, asyncPreRetry, asyncPostRetry);
+						}
+					})
+					.contextWrite(c -> Context.empty())
+		);
 	}
 
 	//===================
diff --git a/reactor-core/src/test/java/reactor/core/publisher/FluxRetryWhenTest.java b/reactor-core/src/test/java/reactor/core/publisher/FluxRetryWhenTest.java
index ceab36edb..1eb0dbad5 100644
--- a/reactor-core/src/test/java/reactor/core/publisher/FluxRetryWhenTest.java
+++ b/reactor-core/src/test/java/reactor/core/publisher/FluxRetryWhenTest.java
@@ -58,6 +58,33 @@ public class FluxRetryWhenTest {
 	Flux<Integer> rangeError = Flux.concat(Flux.range(1, 2),
 			Flux.error(new RuntimeException("forced failure 0")));
 
+	@Test
+	//https://github.com/reactor/reactor-core/issues/3253
+	public void shouldFailWhenOnErrorContinueEnabled() {
+		Mono.create(sink -> {
+					throw new RuntimeException("blah");
+				})
+				.retryWhen(Retry.indefinitely().filter(t -> false))
+				.onErrorContinue((e, o) -> {})
+				.as(StepVerifier::create)
+				.expectError()
+				.verify(Duration.ofSeconds(10));
+	}
+
+	@Test
+	//https://github.com/reactor/reactor-core/issues/3253
+	public void shouldWorkAsExpected() {
+		Mono.just(1)
+			.map(v -> { // ensure original context is propagated
+				throw new RuntimeException("boom");
+			})
+			.retryWhen(Retry.indefinitely().filter(t -> false))
+			.onErrorContinue((e, o) -> {})
+			.as(StepVerifier::create)
+			.expectComplete()
+			.verify(Duration.ofSeconds(10));
+	}
+
 	@Test
 	public void dontRepeat() {
 		AssertSubscriber<Integer> ts = AssertSubscriber.create();