timeout drops an element that races cancellation instead of discarding it #4395

package demo;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeoutException;

import org.junit.jupiter.api.Test;
import reactor.core.publisher.Hooks;
import reactor.test.StepVerifier;
import reactor.test.publisher.TestPublisher;
import reactor.test.subscriber.TestSubscriber;

import static org.assertj.core.api.Assertions.assertThat;

public class TimeoutDiscardTest {

	@Test
	public void discardsElementArrivingAfterCancel() {
		List<Object> discarded = new ArrayList<>();
		TestPublisher<String> source =
				TestPublisher.createNoncompliant(TestPublisher.Violation.DEFER_CANCELLATION);

		TestSubscriber<String> ts = TestSubscriber.builder().initialRequest(1).build();
		source.flux()
		      .timeout(Duration.ofSeconds(30))
		      .doOnDiscard(Object.class, discarded::add)
		      .subscribe(ts);

		ts.cancel();
		source.next("late");

		assertThat(discarded).as("discarded").containsExactly("late");
	}

	@Test
	public void discardsElementArrivingAfterTimeout() {
		List<Object> discarded = new ArrayList<>();
		TestPublisher<String> source =
				TestPublisher.createNoncompliant(TestPublisher.Violation.DEFER_CANCELLATION);

		StepVerifier.withVirtualTime(() -> source.flux()
		                                         .timeout(Duration.ofMillis(100))
		                                         .doOnDiscard(Object.class, discarded::add))
		            .thenAwait(Duration.ofMillis(200))
		            .expectError(TimeoutException.class)
		            .verify(Duration.ofSeconds(5));

		source.next("late");

		assertThat(discarded).as("discarded").containsExactly("late");
	}

	@Test // guard: passes on 3.8.7, and must keep passing
	public void dropsElementArrivingAfterSourceTerminated() {
		List<Object> discarded = new ArrayList<>();
		List<Object> dropped = new ArrayList<>();
		TestPublisher<String> source =
				TestPublisher.createNoncompliant(TestPublisher.Violation.CLEANUP_ON_TERMINATE);

		TestSubscriber<String> ts = TestSubscriber.builder().initialRequest(1).build();
		source.flux()
		      .timeout(Duration.ofSeconds(30))
		      .doOnDiscard(Object.class, discarded::add)
		      .subscribe(ts);

		Hooks.onNextDropped(dropped::add);
		try {
			source.complete();
			source.next("malformed");
		}
		finally {
			Hooks.resetOnNextDropped();
		}

		assertThat(dropped).as("dropped").containsExactly("malformed");
		assertThat(discarded).as("discarded").isEmpty();
	}
}