Further fix async fusion in doFinally|AfterTerminate|onError (#3046)

As a follow-up to #3045, this commit introduces an ArchUnit rule to
surface other cases where a Subscriber is a QueueSubscription but
doesn't implement `onNext` (and thus can break ASYNC fusion).

Three additional case have thus been surfaced, two of which cannot be
fixed entirely.

`doFinally`: in the `poll()` path there was no catching of `qs.poll()`
exceptions BUT unfortunately there is no clean way to implement fused
doFinally with thrown exceptions. For instance, assuming the following
is fused:
```
 .doFinally(sig -> signals.add("finally"))
 .doOnError(e -> signals.add("onError"))
```
then the `doFinally` handler is invoked BEFORE the doOnError (which is
contrary to the documented order and the unfused order). This commit
thus removes `Fuseable` trait from doFinally operators.

`doAfterTerminate`: this is similar to the above situation, but fusion
support needs not be entirely removed from `MonoPeekTerminal` if no
"afterXxx" handler is defined. This commit thus forces fusion to be
negotiated to `NONE` in case a `MonoPeekTerminal` has an
`afterTerminateHandler`, and only in that case.

`doOnError`: didn't catch polling exceptions either, so the `doOnError`
handler from the above snippet was not even invoked. This commit fixes
that case.

Lastly, this commit fixes an out-of-order issue in `StepVerifier`'s
handling of fusion: StepVerifierBuilder must cancel the upstream AFTER
having propagated onError in ASYNC mode, otherwise it may hang.

reactor-core/src/blockHoundTest/java/reactor/core/scheduler/ReactorBlockHoundIntegrationTest.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2019-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2019-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.

reactor-core/src/main/java/reactor/core/publisher/Flux.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2016-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-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.
@@ -4809,9 +4809,6 @@ public final Flux<T> doFirst(Runnable onFirst) {
 	 */
 	public final Flux<T> doFinally(Consumer<SignalType> onFinally) {
 		Objects.requireNonNull(onFinally, "onFinally");
-		if (this instanceof Fuseable) {
-			return onAssembly(new FluxDoFinallyFuseable<>(this, onFinally));
-		}
 		return onAssembly(new FluxDoFinally<>(this, onFinally));
 	}
 

reactor-core/src/main/java/reactor/core/publisher/FluxDoFinally.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2016-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-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.
@@ -45,17 +45,7 @@ final class FluxDoFinally<T> extends InternalFluxOperator<T, T> {
 	final Consumer<SignalType> onFinally;
 
 	@SuppressWarnings("unchecked")
-	static <T> CoreSubscriber<T> createSubscriber(
-			CoreSubscriber<? super T> s, Consumer<SignalType> onFinally,
-			boolean fuseable) {
-
-		if (fuseable) {
-			if(s instanceof ConditionalSubscriber) {
-				return new DoFinallyFuseableConditionalSubscriber<>(
-						(ConditionalSubscriber<?	super T>) s, onFinally);
-			}
-			return new DoFinallyFuseableSubscriber<>(s, onFinally);
-		}
+	static <T> CoreSubscriber<T> createSubscriber(CoreSubscriber<? super T> s, Consumer<SignalType> onFinally) {
 
 		if (s instanceof ConditionalSubscriber) {
 			return new DoFinallyConditionalSubscriber<>((ConditionalSubscriber<? super T>) s, onFinally);
@@ -70,7 +60,7 @@ static <T> CoreSubscriber<T> createSubscriber(
 
 	@Override
 	public CoreSubscriber<? super T> subscribeOrReturn(CoreSubscriber<? super T> actual) {
-		return createSubscriber(actual, onFinally, false);
+		return createSubscriber(actual, onFinally);
 	}
 
 	@Override
@@ -87,15 +77,12 @@ static class DoFinallySubscriber<T> implements InnerOperator<T, T> {
 
 		volatile int once;
 
+		@SuppressWarnings("rawtypes")
 		static final AtomicIntegerFieldUpdater<DoFinallySubscriber> ONCE =
 			AtomicIntegerFieldUpdater.newUpdater(DoFinallySubscriber.class, "once");
 
-		QueueSubscription<T> qs;
-
 		Subscription s;
 
-		boolean syncFused;
-
 		DoFinallySubscriber(CoreSubscriber<? super T> actual, Consumer<SignalType> onFinally) {
 			this.actual = actual;
 			this.onFinally = onFinally;
@@ -112,14 +99,10 @@ public Object scanUnsafe(Attr key) {
 			return InnerOperator.super.scanUnsafe(key);
 		}
 
-		@SuppressWarnings("unchecked")
 		@Override
 		public void onSubscribe(Subscription s) {
 			if (Operators.validate(this.s, s)) {
 				this.s = s;
-				if (s instanceof QueueSubscription) {
-					this.qs = (QueueSubscription<T>)s;
-				}
 
 				actual.onSubscribe(this);
 			}
@@ -175,57 +158,6 @@ public CoreSubscriber<? super T> actual() {
 
 	}
 
-	static class DoFinallyFuseableSubscriber<T> extends DoFinallySubscriber<T>
-		implements Fuseable, QueueSubscription<T> {
-
-		DoFinallyFuseableSubscriber(CoreSubscriber<? super T> actual, Consumer<SignalType> onFinally) {
-			super(actual, onFinally);
-		}
-
-		@Override
-		public int requestFusion(int mode) {
-			QueueSubscription<T> qs = this.qs;
-			if (qs != null && (mode & Fuseable.THREAD_BARRIER) == 0) {
-				int m = qs.requestFusion(mode);
-				if (m != Fuseable.NONE) {
-					syncFused = m == Fuseable.SYNC;
-				}
-				return m;
-			}
-			return Fuseable.NONE;
-		}
-
-		@Override
-		public void clear() {
-			if (qs != null) {
-				qs.clear();
-			}
-		}
-
-		@Override
-		public boolean isEmpty() {
-			return qs == null || qs.isEmpty();
-		}
-
-		@Override
-		@Nullable
-		public T poll() {
-			if (qs == null) {
-				return null;
-			}
-			T v = qs.poll();
-			if (v == null && syncFused) {
-				runFinally(SignalType.ON_COMPLETE);
-			}
-			return v;
-		}
-
-		@Override
-		public int size() {
-			return qs == null ? 0 : qs.size();
-		}
-	}
-
 	static final class DoFinallyConditionalSubscriber<T> extends DoFinallySubscriber<T>
 			implements ConditionalSubscriber<T> {
 
@@ -240,19 +172,4 @@ public boolean tryOnNext(T t) {
 			return ((ConditionalSubscriber<? super T>)actual).tryOnNext(t);
 		}
 	}
-
-	static final class DoFinallyFuseableConditionalSubscriber<T> extends DoFinallyFuseableSubscriber<T>
-		implements ConditionalSubscriber<T> {
-
-		DoFinallyFuseableConditionalSubscriber(ConditionalSubscriber<? super T> actual,
-				Consumer<SignalType> onFinally) {
-			super(actual, onFinally);
-		}
-
-		@Override
-		@SuppressWarnings("unchecked")
-		public boolean tryOnNext(T t) {
-			return ((ConditionalSubscriber<? super T>)actual).tryOnNext(t);
-		}
-	}
 }

reactor-core/src/main/java/reactor/core/publisher/Mono.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2016-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-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.
@@ -2573,9 +2573,6 @@ public final Mono<T> doFirst(Runnable onFirst) {
 	 */
 	public final Mono<T> doFinally(Consumer<SignalType> onFinally) {
 		Objects.requireNonNull(onFinally, "onFinally");
-		if (this instanceof Fuseable) {
-			return onAssembly(new MonoDoFinallyFuseable<>(this, onFinally));
-		}
 		return onAssembly(new MonoDoFinally<>(this, onFinally));
 	}
 

reactor-core/src/main/java/reactor/core/publisher/MonoDoFinally.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2016-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-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.
@@ -44,7 +44,7 @@ final class MonoDoFinally<T> extends InternalMonoOperator<T, T> {
 
 	@Override
 	public CoreSubscriber<? super T> subscribeOrReturn(CoreSubscriber<? super T> actual) {
-		return FluxDoFinally.createSubscriber(actual, onFinally, false);
+		return FluxDoFinally.createSubscriber(actual, onFinally);
 	}
 
 	@Override

reactor-core/src/main/java/reactor/core/publisher/MonoPeekTerminal.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2016-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-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.
@@ -321,7 +321,22 @@ public CoreSubscriber<? super T> actual() {
 		public T poll() {
 			assert queueSubscription != null;
 			boolean d = done;
-			T v = queueSubscription.poll();
+			T v;
+			try {
+				v = queueSubscription.poll();
+			}
+			catch (Throwable pe) {
+				if (parent.onErrorCall != null) {
+					try {
+						parent.onErrorCall.accept(pe);
+					}
+					catch (Throwable t) {
+						t = Operators.onOperatorError(null, pe, t, actual.currentContext());
+						throw Exceptions.propagate(t);
+					}
+				}
+				throw pe;
+			}
 			if (!valued && (v != null || d || sourceMode == SYNC)) {
 				valued = true;
 				//TODO include onEmptyCall here as well?
@@ -334,16 +349,8 @@ public T poll() {
 								actual.currentContext()));
 					}
 				}
-				if (parent.onAfterTerminateCall != null) {
-					try {
-						parent.onAfterTerminateCall.accept(v, null);
-					}
-					catch (Throwable t) {
-						Operators.onErrorDropped(Operators.onOperatorError(t,
-								actual.currentContext()),
-								actual.currentContext());
-					}
-				}
+				//if parent.onAfterTerminateCall is set, fusion MUST be negotiated to NONE
+				//because there's no way to correctly support onAfterError in the poll() scenario
 			}
 			return v;
 		}
@@ -362,7 +369,13 @@ public void clear() {
 		@Override
 		public int requestFusion(int requestedMode) {
 			int m;
-			if (queueSubscription == null) { //source wasn't actually Fuseable
+			if (queueSubscription == null || parent.onAfterTerminateCall != null) {
+				/*
+				Two cases where the configuration doesn't allow fusion:
+				 - source wasn't actually Fuseable
+				 - onAfterTerminateCall is set (which cannot be correctly implemented in the case
+				   qs.poll() throws)
+				 */
 				m = NONE;
 			}
 			else if ((requestedMode & THREAD_BARRIER) != 0) {