3.x: Fix delay-cancellation race producing random subsequent emissions

src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableDelay.java

@@ -111,7 +111,9 @@ final class OnNext implements Runnable {
 
             @Override
             public void run() {
-                downstream.onNext(t);
+                if (!w.isDisposed()) {
+                    downstream.onNext(t);
+                }
             }
         }
 

src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableDelay.java

@@ -111,7 +111,9 @@ final class OnNext implements Runnable {
 
             @Override
             public void run() {
-                downstream.onNext(t);
+                if (!w.isDisposed()) {
+                    downstream.onNext(t);
+                }
             }
         }
 

src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableDelayTest.java

@@ -20,6 +20,7 @@
 import java.util.*;
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.*;
+import java.util.concurrent.locks.LockSupport;
 
 import org.junit.*;
 import org.mockito.InOrder;
@@ -28,6 +29,7 @@
 import io.reactivex.rxjava3.core.*;
 import io.reactivex.rxjava3.exceptions.TestException;
 import io.reactivex.rxjava3.functions.*;
+import io.reactivex.rxjava3.internal.disposables.SequentialDisposable;
 import io.reactivex.rxjava3.internal.functions.Functions;
 import io.reactivex.rxjava3.processors.PublishProcessor;
 import io.reactivex.rxjava3.schedulers.*;
@@ -1030,4 +1032,38 @@ public Publisher<Object> apply(Integer t) throws Exception {
         .to(TestHelper.<Integer>testConsumer())
         .assertFailureAndMessage(NullPointerException.class, "The itemDelay returned a null Publisher");
     }
+
+    @Test
+    public void cancelShouldPreventRandomSubsequentEmissions() {
+        for (int attempt = 1; attempt < 100; attempt ++) {
+
+            SequentialDisposable disposable = new SequentialDisposable();
+            ConcurrentLinkedQueue<Integer> sink = new ConcurrentLinkedQueue<>();
+
+            disposable.replace(
+                Flowable.range(1, 10)
+                .delay(1, TimeUnit.MICROSECONDS, Schedulers.computation(), true)
+                .doOnNext(v -> {
+                    if (v == 1) {
+                        Schedulers.computation().scheduleDirect(disposable::dispose);
+                    }
+                    sink.offer(v);
+                })
+            .subscribe());
+
+            LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(1));
+
+            Integer last = null;
+
+            while (!sink.isEmpty()) {
+                Integer current = sink.poll();
+
+                if (last != null && last + 1 != current) {
+                    fail("Emission hole: " + last + " -> " + current);
+                }
+
+                last = current;
+            }
+        }
+    }
 }

src/test/java/io/reactivex/rxjava3/internal/operators/observable/ObservableDelayTest.java

@@ -20,6 +20,7 @@
 import java.util.*;
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.LockSupport;
 
 import org.junit.*;
 import org.mockito.InOrder;
@@ -29,6 +30,7 @@
 import io.reactivex.rxjava3.core.Observer;
 import io.reactivex.rxjava3.exceptions.TestException;
 import io.reactivex.rxjava3.functions.*;
+import io.reactivex.rxjava3.internal.disposables.SequentialDisposable;
 import io.reactivex.rxjava3.internal.functions.Functions;
 import io.reactivex.rxjava3.observers.*;
 import io.reactivex.rxjava3.schedulers.*;
@@ -978,4 +980,37 @@ public Observable<Object> apply(Integer t) throws Exception {
         .to(TestHelper.<Integer>testConsumer())
         .assertFailureAndMessage(NullPointerException.class, "The itemDelay returned a null ObservableSource");
     }
-}
+
+    @Test
+    public void cancelShouldPreventRandomSubsequentEmissions() {
+        for (int attempt = 1; attempt < 100; attempt ++) {
+
+            SequentialDisposable disposable = new SequentialDisposable();
+            ConcurrentLinkedQueue<Integer> sink = new ConcurrentLinkedQueue<>();
+
+            disposable.replace(
+                Observable.range(1, 10)
+                .delay(1, TimeUnit.MICROSECONDS, Schedulers.computation(), true)
+                .doOnNext(v -> {
+                    if (v == 1) {
+                        Schedulers.computation().scheduleDirect(disposable::dispose);
+                    }
+                    sink.offer(v);
+                })
+            .subscribe());
+
+            LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(1));
+
+            Integer last = null;
+
+            while (!sink.isEmpty()) {
+                Integer current = sink.poll();
+
+                if (last != null && last + 1 != current) {
+                    fail("Emission hole: " + last + " -> " + current);
+                }
+
+                last = current;
+            }
+        }
+    }}