fix reactor#2042 collect() discards on consumer error when fused

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

@@ -1566,10 +1566,9 @@ public MonoSubscriber(CoreSubscriber<? super O> actual) {
 		@Override
 		public void cancel() {
 			O v = value;
-			if (STATE.getAndSet(this, CANCELLED) <= HAS_REQUEST_NO_VALUE) {
-				discard(v);
-			}
 			value = null;
+			STATE.set(this, CANCELLED);
+			discard(v);
 		}
 
 		@Override

reactor-core/src/test/java/reactor/core/publisher/MonoCollectTest.java

@@ -27,6 +27,7 @@
 import org.junit.Test;
 import org.reactivestreams.Subscription;
 import reactor.core.CoreSubscriber;
+import reactor.core.Fuseable;
 import reactor.core.Scannable;
 import reactor.test.StepVerifier;
 import reactor.test.subscriber.AssertSubscriber;
@@ -154,6 +155,39 @@ public void discardElementOnAccumulatorFailure() {
 		    .hasDiscardedExactly(1);
 	}
 
+	@Test
+	public void discardElementAndBufferOnAccumulatorLateFailure() {
+		Flux.just(1, 2, 3, 4)
+		    .hide()
+		    .collect(ArrayList::new, (l, t) -> {
+			    if (t == 3) {
+				    throw new IllegalStateException("accumulator: boom");
+			    }
+			    l.add(t);
+		    })
+		    .as(StepVerifier::create)
+		    .expectErrorMessage("accumulator: boom")
+		    .verifyThenAssertThat()
+		    .hasDiscardedExactly(1, 2, 3);
+	}
+
+	@Test
+	public void discardElementAndBufferOnAccumulatorLateFailure_fused() {
+		Flux.just(1, 2, 3, 4)
+		    .collect(ArrayList::new, (l, t) -> {
+			    if (t == 3) {
+				    throw new IllegalStateException("accumulator: boom");
+			    }
+			    l.add(t);
+		    })
+		    .as(StepVerifier::create)
+		    //WARNING: we need to request fusion so this expectFusion is important
+		    .expectFusion(Fuseable.ASYNC)
+		    .expectErrorMessage("accumulator: boom")
+		    .verifyThenAssertThat()
+		    .hasDiscardedExactly(1, 2, 3);
+	}
+
 	@Test
 	public void discardListElementsOnError() {
 		Mono<List<Integer>> test =