http-netty: fix JavaNetSoTimeoutHttpConnectionFilter leak

servicetalk-http-netty/src/test/java/io/servicetalk/http/netty/JavaNetSoTimeoutHttpConnectionFilterTest.java

@@ -21,6 +21,7 @@
 import io.servicetalk.concurrent.Cancellable;
 import io.servicetalk.concurrent.api.DelegatingExecutor;
 import io.servicetalk.concurrent.api.Executor;
+import io.servicetalk.concurrent.api.ExecutorExtension;
 import io.servicetalk.concurrent.api.Publisher;
 import io.servicetalk.concurrent.api.Single;
 import io.servicetalk.concurrent.api.TestExecutor;
@@ -67,6 +68,7 @@
 import javax.annotation.Nullable;
 
 import static io.servicetalk.buffer.netty.BufferAllocators.DEFAULT_ALLOCATOR;
+import static io.servicetalk.concurrent.api.ExecutorExtension.withTestExecutor;
 import static io.servicetalk.concurrent.api.Publisher.from;
 import static io.servicetalk.concurrent.internal.TestTimeoutConstants.CI;
 import static io.servicetalk.context.api.ContextMap.Key.newKey;
@@ -83,7 +85,6 @@
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.is;
-import static org.hamcrest.Matchers.isA;
 import static org.hamcrest.Matchers.nullValue;
 import static org.hamcrest.Matchers.startsWith;
 import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -92,6 +93,9 @@
 
 class JavaNetSoTimeoutHttpConnectionFilterTest {
 
+    @RegisterExtension
+    static final ExecutorExtension<TestExecutor> testExecutorExtension = withTestExecutor().setClassLevel(true);
+
     @RegisterExtension
     static final ExecutionContextExtension SERVER_CTX =
             ExecutionContextExtension.cached("server-io", "server-executor")
@@ -114,12 +118,11 @@ class JavaNetSoTimeoutHttpConnectionFilterTest {
     private static ServerContext server;
     @Nullable
     private static BlockingHttpClient client;
-
-    @Nullable
-    private static TestExecutor timeoutExecutor;
+    private static TestExecutor testExecutor;
 
     @BeforeAll
     static void setUp() throws Exception {
+        testExecutor = testExecutorExtension.executor();
         server = newServerBuilder(SERVER_CTX).listenStreamingAndAwait((ctx, request, responseFactory) -> {
             Buffer hello = ctx.executionContext().bufferAllocator().fromAscii("Hello");
 
@@ -151,8 +154,6 @@ static void setUp() throws Exception {
                 .appendConnectionFilter(new JavaNetSoTimeoutHttpConnectionFilter(
                         (metaData, timeSource) -> metaData.context().get(READ_TIMEOUT_KEY)))
                 .buildBlocking();
-
-        timeoutExecutor = new TestExecutor();
     }
 
     @AfterAll
@@ -317,15 +318,14 @@ void negativeTimeout() {
     void racingResponsesAreCleanedUp() {
         AtomicBoolean isCancelled = new AtomicBoolean();
         TestSingle<StreamingHttpResponse> responseSingle = new TestSingle<>();
-        final Duration timeout = Duration.ofMillis(100);
-        Future<StreamingHttpResponse> result = applyFilter(timeout, responseSingle
+        Future<StreamingHttpResponse> result = applyFilter(READ_TIMEOUT_VALUE, responseSingle
                 .whenCancel(() -> isCancelled.set(true))).toFuture();
 
-        assertThat(responseSingle.isSubscribed(), is(true));
-        testExecutor().advanceTimeBy(100, TimeUnit.MILLISECONDS);
+        responseSingle.awaitSubscribed();
+        testExecutor.advanceTimeBy(READ_TIMEOUT_VALUE.toMillis(), TimeUnit.MILLISECONDS);
 
-        ExecutionException ex = assertThrows(ExecutionException.class, () -> result.get());
-        assertThat(ex.getCause(), isA(SocketTimeoutException.class));
+        ExecutionException ex = assertThrows(ExecutionException.class, result::get);
+        assertThat(ex.getCause(), is(instanceOf(SocketTimeoutException.class)));
 
         // the response should have been cancelled.
         assertThat(isCancelled.get(), is(true));
@@ -341,29 +341,31 @@ void racingResponsesAreCleanedUp() {
     @Test
     void timerIsCancelledOnSuccessfulResponse() throws Exception {
         TestSingle<StreamingHttpResponse> responseSingle = new TestSingle<>();
-        final Duration timeout = Duration.ofMillis(100);
-        Future<StreamingHttpResponse> responseFuture = applyFilter(timeout, responseSingle).toFuture();
+        Future<StreamingHttpResponse> responseFuture = applyFilter(READ_TIMEOUT_VALUE, responseSingle).toFuture();
         responseSingle.awaitSubscribed();
 
-        responseSingle.onSuccess(responseRawWith(Publisher.empty()));
-        responseFuture.get();
-        assertThat(testExecutor().scheduledTasksPending(), is(0));
+        assertThat(testExecutor.scheduledTasksPending(), is(1));
+        StreamingHttpResponse response = responseRawWith(Publisher.empty());
+        responseSingle.onSuccess(response);
+        assertThat(responseFuture.get(), is(response));
+        assertThat(testExecutor.scheduledTasksPending(), is(0));
     }
 
     @Test
     void timerLosingRaceDoesntTriggerRequestCancellation() throws Exception {
         TestSingle<StreamingHttpResponse> responseSingle = new TestSingle<>();
-        final Duration timeout = Duration.ofMillis(100);
         AtomicBoolean responseCancelled = new AtomicBoolean();
-        Future<StreamingHttpResponse> responseFuture = applyFilter(timeout, responseSingle
+        Future<StreamingHttpResponse> responseFuture = applyFilter(READ_TIMEOUT_VALUE, responseSingle
                 .whenCancel(() -> responseCancelled.set(true)), true).toFuture();
         responseSingle.awaitSubscribed();
 
-        responseSingle.onSuccess(responseRawWith(Publisher.empty()));
-        responseFuture.get();
-        assertThat(testExecutor().scheduledTasksPending(), is(1));
-        testExecutor().advanceTimeBy(100, TimeUnit.MILLISECONDS);
-        assertThat(testExecutor().scheduledTasksPending(), is(0));
+        StreamingHttpResponse response = responseRawWith(Publisher.empty());
+        responseSingle.onSuccess(response);
+        assertThat(responseFuture.get(), is(response));
+        // we use ignoreCancel == true for TestExecutor to simulate that timeout may race with response onSuccess
+        assertThat(testExecutor.scheduledTasksPending(), is(1));
+        testExecutor.advanceTimeBy(READ_TIMEOUT_VALUE.toMillis(), TimeUnit.MILLISECONDS);
+        assertThat(testExecutor.scheduledTasksPending(), is(0));
 
         assertThat(responseCancelled.get(), is(false));
     }
@@ -374,9 +376,8 @@ void upstreamCancellationIsAlwaysPropagated() throws Exception {
         // if a response triggers before upstream cancellation, the upstream cancellation would not be propagated.
         AtomicBoolean isCancelled = new AtomicBoolean();
         TestSingle<StreamingHttpResponse> responseSingle = new TestSingle<>();
-        final Duration timeout = Duration.ofMillis(100);
         CountDownLatch responseReceived = new CountDownLatch(1);
-        Cancellable cancellable = applyFilter(timeout, responseSingle)
+        Cancellable cancellable = applyFilter(READ_TIMEOUT_VALUE, responseSingle)
                 .whenOnSuccess(resp -> responseReceived.countDown())
                 .toCompletable().subscribe();
 
@@ -395,7 +396,6 @@ private static Single<StreamingHttpResponse> applyFilter(Duration timeout, Singl
     private static Single<StreamingHttpResponse> applyFilter(Duration timeout, Single<StreamingHttpResponse> response,
                                                              boolean ignoreCancel) {
         FilterableStreamingHttpConnection connection = mock(FilterableStreamingHttpConnection.class);
-        // TODO: how do I consume the request?
         ArgumentCaptor<StreamingHttpRequest> requestCaptor = ArgumentCaptor.forClass(StreamingHttpRequest.class);
         when(connection.request(requestCaptor.capture())).thenAnswer(new Answer<Object>() {
             @Override
@@ -405,7 +405,7 @@ public Object answer(InvocationOnMock invocation) throws Throwable {
             }
         });
 
-        Executor exec = ignoreCancel ? new DelegatingExecutor(testExecutor()) {
+        Executor exec = ignoreCancel ? new DelegatingExecutor(testExecutor) {
             @Override
             public Cancellable schedule(Runnable task, long delay, TimeUnit unit) throws RejectedExecutionException {
                 super.schedule(task, delay, unit);
@@ -417,17 +417,12 @@ public Cancellable schedule(Runnable task, Duration delay) throws RejectedExecut
                 super.schedule(task, delay);
                 return Cancellable.IGNORE_CANCEL;
             }
-        } : testExecutor();
+        } : testExecutor;
 
         return new JavaNetSoTimeoutHttpConnectionFilter(timeout, exec).create(connection)
                 .request(newRequest().toStreamingRequest());
     }
 
-    private static TestExecutor testExecutor() {
-        assert timeoutExecutor != null;
-        return timeoutExecutor;
-    }
-
     private static BlockingHttpClient client() {
         assert client != null;
         return client;

servicetalk-http-utils/src/main/java/io/servicetalk/http/utils/JavaNetSoTimeoutHttpConnectionFilter.java

@@ -42,6 +42,9 @@
 import io.servicetalk.http.utils.AbstractTimeoutHttpFilter.FixedDuration;
 import io.servicetalk.transport.api.ExecutionContext;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import java.net.SocketOptions;
 import java.net.SocketTimeoutException;
 import java.time.Duration;
@@ -51,6 +54,7 @@
 import javax.annotation.Nullable;
 
 import static io.servicetalk.concurrent.api.SourceAdapters.toSource;
+import static io.servicetalk.utils.internal.ThrowableUtils.throwException;
 import static java.util.Objects.requireNonNull;
 
 /**
@@ -79,6 +83,8 @@
  */
 public final class JavaNetSoTimeoutHttpConnectionFilter implements StreamingHttpConnectionFilterFactory {
 
+    private static final Logger LOGGER = LoggerFactory.getLogger(JavaNetSoTimeoutHttpConnectionFilter.class);
+
     private final BiFunction<HttpRequestMetaData, TimeSource, Duration> timeoutForRequest;
     @Nullable
     private final Executor timeoutExecutor;
@@ -189,36 +195,57 @@ static final class RequestTimeoutSubscriber implements Subscriber<StreamingHttpR
                 AtomicIntegerFieldUpdater.newUpdater(RequestTimeoutSubscriber.class, "once");
 
         private final DelayedCancellable requestCancellable = new DelayedCancellable();
-        private final DelayedCancellable timeoutCancellable = new DelayedCancellable();
+        private final Cancellable timeoutCancellable;
         private final Subscriber<? super StreamingHttpResponse> delegate;
 
-        private final Completable requestComplete;
         private final Duration timeout;
         private final Executor timeoutExecutor;
         @SuppressWarnings("unused")
         private volatile int once;
 
         RequestTimeoutSubscriber(Subscriber<? super StreamingHttpResponse> delegate, Completable requestComplete,
-                                        Duration timeout, Executor timeoutExecutor) {
+                                 Duration timeout, Executor timeoutExecutor) {
             this.delegate = delegate;
-            this.requestComplete = requestComplete;
             this.timeout = timeout;
             this.timeoutExecutor = timeoutExecutor;
+            timeoutCancellable = requestComplete.concat(Completable.never()
+                    .timeout(timeout, timeoutExecutor)).beforeOnError(this::handleInterruptions).subscribe();
         }
 
         @Override
         public void onSubscribe(Cancellable cancellable) {
-            delegate.onSubscribe(() -> {
-                // We don't need to condition cancellation here on the `once()` call because the `DelayedCancellable`
-                // will already enforce idempotency of the cancel call, and it's fine if we're racing cancels with the
-                // results since cleanup is gated on the `once()` call.
-                once();
-                timeoutCancellable.cancel();
-                requestCancellable.cancel();
-            });
-            requestCancellable.delayedCancellable(cancellable);
-            timeoutCancellable.delayedCancellable(requestComplete.concat(Completable.never()
-                    .timeout(timeout, timeoutExecutor)).beforeOnError(this::handleInterruptions).subscribe());
+            try {
+                delegate.onSubscribe(() -> {
+                    once();
+                    Throwable t = null;
+                    try {
+                        timeoutCancellable.cancel();
+                    } catch (Throwable tt) {
+                        t = tt;
+                    }
+                    try {
+                        requestCancellable.cancel();
+                    } catch (Throwable tt) {
+                        if (t == null) {
+                            t = tt;
+                        } else {
+                            t.addSuppressed(tt);
+                        }
+                    }
+                    if (t != null) {
+                        throwException(t);
+                    }
+                });
+                requestCancellable.delayedCancellable(cancellable);
+            } catch (Throwable cause) {
+                try {
+                    cancellable.cancel();
+                } catch (Throwable t) {
+                    cause.addSuppressed(t);
+                }
+                onError(cause);
+                LOGGER.warn("Unexpected exception from onSubscribe of Subscriber {}.", delegate, cause);
+            }
         }
 
         private void handleInterruptions(Throwable t) {
@@ -227,10 +254,8 @@ private void handleInterruptions(Throwable t) {
                 Throwable result = t;
                 // We can get a SocketTimeoutException waiting for a 100 Continue response.
                 if (t instanceof TimeoutException) {
-                result = newStacklessSocketTimeoutException(
-                        "Read timed out after " + timeout.toMillis() +
-                                "ms waiting for response meta-data")
-                        .initCause(t);
+                    result = newStacklessSocketTimeoutException("Read timed out after " + timeout.toMillis() +
+                            "ms waiting for response meta-data").initCause(t);
                 }
                 delegate.onError(result);
             }
@@ -258,13 +283,18 @@ public void onSuccess(@Nullable StreamingHttpResponse result) {
         @Override
         public void onError(Throwable t) {
             if (once()) {
-                timeoutCancellable.cancel();
+                try {
+                    timeoutCancellable.cancel();
+                } catch (Throwable tt) {
+                    delegate.onError(tt);
+                    return;
+                }
                 delegate.onError(t);
             }
         }
 
         private boolean once() {
-            return 0 == onceUpdater.getAndSet(this, 1);
+            return onceUpdater.compareAndSet(this, 0, 1);
         }
     }