8208693: HttpClient: Extend the request timeout's scope to cover the response body

Reviewed-by: jpai, dfuchs

Author: vy

src/java.net.http/share/classes/java/net/http/HttpClient.java

@@ -312,10 +312,22 @@ public interface Builder {
          * need to be established, for example if a connection can be reused
          * from a previous request, then this timeout duration has no effect.
          *
+         * @implSpec
+         * A connection timeout applies to the entire connection phase, from the
+         * moment a connection is requested until it is established.
+         * Implementations are recommended to ensure that the connection timeout
+         * covers any SSL/TLS handshakes.
+         *
+         * @implNote
+         * The built-in JDK implementation of the connection timeout covers any
+         * SSL/TLS handshakes.
+         *
          * @param duration the duration to allow the underlying connection to be
          *                 established
          * @return this builder
          * @throws IllegalArgumentException if the duration is non-positive
+         * @see HttpRequest.Builder#timeout(Duration) Configuring timeout for
+         * request execution
          */
         public Builder connectTimeout(Duration duration);
 

src/java.net.http/share/classes/java/net/http/HttpRequest.java

@@ -258,12 +258,28 @@ public interface Builder {
          * {@link HttpClient#sendAsync(java.net.http.HttpRequest,
          * java.net.http.HttpResponse.BodyHandler) HttpClient::sendAsync}
          * completes exceptionally with an {@code HttpTimeoutException}. The effect
-         * of not setting a timeout is the same as setting an infinite Duration,
-         * i.e. block forever.
+         * of not setting a timeout is the same as setting an infinite
+         * {@code Duration}, i.e., block forever.
+         *
+         * @implSpec
+         * A timeout applies to the duration measured from the instant the
+         * request execution starts to, <em>at least</em>, the instant an
+         * {@link HttpResponse} is constructed. The elapsed time includes
+         * obtaining a connection for transport and retrieving the response
+         * headers.
+         *
+         * @implNote
+         * The JDK built-in implementation applies timeout over the duration
+         * measured from the instant the request execution starts to <b>the
+         * instant the response body is consumed</b>, if present. This is
+         * implemented by stopping the timer after the response body subscriber
+         * completion.
          *
          * @param duration the timeout duration
          * @return this builder
          * @throws IllegalArgumentException if the duration is non-positive
+         * @see HttpClient.Builder#connectTimeout(Duration) Configuring
+         * timeout for connection establishment
          */
         public abstract Builder timeout(Duration duration);
 

src/java.net.http/share/classes/java/net/http/WebSocket.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2015, 2024, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2015, 2025, Oracle and/or its affiliates. All rights reserved.
  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
  *
  * This code is free software; you can redistribute it and/or modify it
@@ -144,6 +144,16 @@ interface Builder {
          * {@link HttpTimeoutException}. If this method is not invoked then the
          * infinite timeout is assumed.
          *
+         * @implSpec
+         * A connection timeout applies to the entire connection phase, from the
+         * moment a connection is requested until it is established.
+         * Implementations are recommended to ensure that the connection timeout
+         * covers any WebSocket and SSL/TLS handshakes.
+         *
+         * @implNote
+         * The built-in JDK implementation of the connection timeout covers any
+         * WebSocket and SSL/TLS handshakes.
+         *
          * @param timeout
          *         the timeout, non-{@linkplain Duration#isNegative() negative},
          *         non-{@linkplain Duration#ZERO ZERO}

src/java.net.http/share/classes/jdk/internal/net/http/ExchangeImpl.java

@@ -581,6 +581,18 @@ void nullBody(HttpResponse<T> resp, Throwable t) {
         // Needed for HTTP/2 to subscribe a dummy subscriber and close the stream
     }
 
+    /**
+     * {@return {@code true}, if it is allowed to cancel the request timer on
+     * response body subscriber termination; {@code false}, otherwise}
+     *
+     * @param webSocket indicates if the associated request is a WebSocket handshake
+     * @param statusCode the status code of the associated response
+     */
+    static boolean cancelTimerOnResponseBodySubscriberTermination(
+            boolean webSocket, int statusCode) {
+        return webSocket || statusCode < 100 || statusCode >= 200;
+    }
+
     /* The following methods have separate HTTP/1.1 and HTTP/2 implementations */
 
     abstract CompletableFuture<ExchangeImpl<T>> sendHeadersAsync();

src/java.net.http/share/classes/jdk/internal/net/http/Http1Exchange.java

@@ -206,8 +206,15 @@ private void error() {
      */
     static final class Http1ResponseBodySubscriber<U> extends HttpBodySubscriberWrapper<U> {
         final Http1Exchange<U> exchange;
-        Http1ResponseBodySubscriber(BodySubscriber<U> userSubscriber, Http1Exchange<U> exchange) {
+
+        private final boolean cancelTimerOnTermination;
+
+        Http1ResponseBodySubscriber(
+                BodySubscriber<U> userSubscriber,
+                boolean cancelTimerOnTermination,
+                Http1Exchange<U> exchange) {
             super(userSubscriber);
+            this.cancelTimerOnTermination = cancelTimerOnTermination;
             this.exchange = exchange;
         }
 
@@ -220,6 +227,14 @@ protected void register() {
         protected void unregister() {
             exchange.unregisterResponseSubscriber(this);
         }
+
+        @Override
+        protected void onTermination() {
+            if (cancelTimerOnTermination) {
+                exchange.exchange.multi.cancelTimer();
+            }
+        }
+
     }
 
     @Override
@@ -459,9 +474,10 @@ CompletableFuture<T> readBodyAsync(BodyHandler<T> handler,
     @Override
     Http1ResponseBodySubscriber<T> createResponseSubscriber(BodyHandler<T> handler, ResponseInfo response) {
         BodySubscriber<T> subscriber = handler.apply(response);
-        Http1ResponseBodySubscriber<T> bs =
-                new Http1ResponseBodySubscriber<T>(subscriber, this);
-        return bs;
+        var cancelTimerOnTermination =
+                cancelTimerOnResponseBodySubscriberTermination(
+                        exchange.request().isWebSocket(), response.statusCode());
+        return new Http1ResponseBodySubscriber<>(subscriber, cancelTimerOnTermination, this);
     }
 
     @Override

src/java.net.http/share/classes/jdk/internal/net/http/Http3ExchangeImpl.java

@@ -554,8 +554,12 @@ private void unregisterResponseSubscriber(Http3StreamResponseSubscriber<?> subsc
     }
 
     final class Http3StreamResponseSubscriber<U> extends HttpBodySubscriberWrapper<U> {
-        Http3StreamResponseSubscriber(BodySubscriber<U> subscriber) {
+
+        private final boolean cancelTimerOnTermination;
+
+        Http3StreamResponseSubscriber(BodySubscriber<U> subscriber, boolean cancelTimerOnTermination) {
             super(subscriber);
+            this.cancelTimerOnTermination = cancelTimerOnTermination;
         }
 
         @Override
@@ -568,6 +572,13 @@ protected void register() {
             registerResponseSubscriber(this);
         }
 
+        @Override
+        protected void onTermination() {
+            if (cancelTimerOnTermination) {
+                exchange.multi.cancelTimer();
+            }
+        }
+
         @Override
         protected void logComplete(Throwable error) {
             if (error == null) {
@@ -590,9 +601,10 @@ protected void logComplete(Throwable error) {
     Http3StreamResponseSubscriber<T> createResponseSubscriber(BodyHandler<T> handler,
                                                               ResponseInfo response) {
         if (debug.on()) debug.log("Creating body subscriber");
-        Http3StreamResponseSubscriber<T> subscriber =
-                new Http3StreamResponseSubscriber<>(handler.apply(response));
-        return subscriber;
+        var cancelTimerOnTermination =
+                cancelTimerOnResponseBodySubscriberTermination(
+                        exchange.request().isWebSocket(), response.statusCode());
+        return new Http3StreamResponseSubscriber<>(handler.apply(response), cancelTimerOnTermination);
     }
 
     @Override

src/java.net.http/share/classes/jdk/internal/net/http/HttpClientImpl.java

@@ -1880,6 +1880,13 @@ void cancelTimer(TimeoutEvent event) {
         }
     }
 
+    // Visible for tests
+    List<TimeoutEvent> timers() {
+        synchronized (this) {
+            return new ArrayList<>(timeouts);
+        }
+    }
+
     /**
      * Purges ( handles ) timer events that have passed their deadline, and
      * returns the amount of time, in milliseconds, until the next earliest

src/java.net.http/share/classes/jdk/internal/net/http/MultiExchange.java

@@ -25,7 +25,6 @@
 
 package jdk.internal.net.http;
 
-import java.io.IOError;
 import java.io.IOException;
 import java.lang.ref.WeakReference;
 import java.net.ConnectException;
@@ -254,7 +253,7 @@ public Optional<Duration> remainingConnectTimeout() {
                 .map(ConnectTimeoutTracker::getRemaining);
     }
 
-    private void cancelTimer() {
+    void cancelTimer() {
         if (responseTimerEvent != null) {
             client.cancelTimer(responseTimerEvent);
             responseTimerEvent = null;
@@ -404,6 +403,8 @@ private HttpResponse<T> setNewResponse(HttpRequest request, Response r, T body,
                         processAltSvcHeader(r, client(), currentreq);
                         Exchange<T> exch = getExchange();
                         if (bodyNotPermitted(r)) {
+                            // No response body consumption is expected, we can cancel the timer right away
+                            cancelTimer();
                             if (bodyIsPresent(r)) {
                                 IOException ioe = new IOException(
                                     "unexpected content length header with 204 response");
@@ -467,6 +468,8 @@ private CompletableFuture<Response> retryRequest() {
 
     private CompletableFuture<Response> responseAsyncImpl(final boolean applyReqFilters) {
         if (currentreq.timeout().isPresent()) {
+            // Retried/Forwarded requests should reset the timer, if present
+            cancelTimer();
             responseTimerEvent = ResponseTimerEvent.of(this);
             client.registerTimer(responseTimerEvent);
         }
@@ -502,7 +505,6 @@ private CompletableFuture<Response> responseAsyncImpl(final boolean applyReqFilt
                         }
                         return completedFuture(response);
                     } else {
-                        cancelTimer();
                         setNewResponse(currentreq, response, null, exch);
                         if (currentreq.isWebSocket()) {
                             // need to close the connection and open a new one.
@@ -520,11 +522,18 @@ private CompletableFuture<Response> responseAsyncImpl(final boolean applyReqFilt
                     } })
                 .handle((response, ex) -> {
                     // 5. handle errors and cancel any timer set
-                    cancelTimer();
                     if (ex == null) {
                         assert response != null;
                         return completedFuture(response);
                     }
+
+                    // Cancel the timer. Note that we only do so if the
+                    // response has completed exceptionally. That is, we don't
+                    // cancel the timer if there are no exceptions, since the
+                    // response body might still get consumed, and it is
+                    // still subject to the response timer.
+                    cancelTimer();
+
                     // all exceptions thrown are handled here
                     final RetryContext retryCtx = checkRetryEligible(ex, exch);
                     assert retryCtx != null : "retry context is null";

src/java.net.http/share/classes/jdk/internal/net/http/Stream.java

@@ -390,9 +390,10 @@ CompletableFuture<T> readBodyAsync(HttpResponse.BodyHandler<T> handler,
 
     @Override
     Http2StreamResponseSubscriber<T> createResponseSubscriber(BodyHandler<T> handler, ResponseInfo response) {
-        Http2StreamResponseSubscriber<T> subscriber =
-                new Http2StreamResponseSubscriber<>(handler.apply(response));
-        return subscriber;
+        var cancelTimerOnTermination =
+                cancelTimerOnResponseBodySubscriberTermination(
+                        exchange.request().isWebSocket(), response.statusCode());
+        return new Http2StreamResponseSubscriber<>(handler.apply(response), cancelTimerOnTermination);
     }
 
     // The Http2StreamResponseSubscriber is registered with the HttpClient
@@ -1694,6 +1695,11 @@ CompletableFuture<T> readBodyAsync(
                         .whenComplete((v, t) -> pushGroup.pushError(t));
         }
 
+        @Override
+        Http2StreamResponseSubscriber<T> createResponseSubscriber(BodyHandler<T> handler, ResponseInfo response) {
+            return new Http2StreamResponseSubscriber<T>(handler.apply(response), false);
+        }
+
         @Override
         void completeResponse(Response r) {
             Log.logResponse(r::toString);
@@ -1924,8 +1930,12 @@ public void onMaxHeaderListSizeReached(long size, int maxHeaderListSize) throws
     }
 
     final class Http2StreamResponseSubscriber<U> extends HttpBodySubscriberWrapper<U> {
-        Http2StreamResponseSubscriber(BodySubscriber<U> subscriber) {
+
+        private final boolean cancelTimerOnTermination;
+
+        Http2StreamResponseSubscriber(BodySubscriber<U> subscriber, boolean cancelTimerOnTermination) {
             super(subscriber);
+            this.cancelTimerOnTermination = cancelTimerOnTermination;
         }
 
         @Override
@@ -1938,6 +1948,13 @@ protected void unregister() {
             unregisterResponseSubscriber(this);
         }
 
+        @Override
+        protected void onTermination() {
+            if (cancelTimerOnTermination) {
+                exchange.multi.cancelTimer();
+            }
+        }
+
     }
 
     private static final VarHandle DEREGISTERED;

src/java.net.http/share/classes/jdk/internal/net/http/common/HttpBodySubscriberWrapper.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2022, 2023, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2022, 2025, Oracle and/or its affiliates. All rights reserved.
  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
  *
  * This code is free software; you can redistribute it and/or modify it
@@ -33,7 +33,6 @@
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.Flow;
 import java.util.concurrent.Flow.Subscription;
-import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.locks.ReentrantLock;
 
@@ -51,7 +50,6 @@ public class HttpBodySubscriberWrapper<T> implements TrustedSubscriber<T> {
     public static final Comparator<HttpBodySubscriberWrapper<?>> COMPARE_BY_ID
             = Comparator.comparing(HttpBodySubscriberWrapper::id);
 
-
     public static final Flow.Subscription NOP = new Flow.Subscription() {
         @Override
         public void request(long n) { }
@@ -75,7 +73,18 @@ public HttpBodySubscriberWrapper(BodySubscriber<T> userSubscriber) {
         this.userSubscriber = userSubscriber;
     }
 
-    private class SubscriptionWrapper implements Subscription {
+    /**
+     * A callback to be invoked <em>before</em> termination, whether due to the
+     * completion or failure of the subscriber, or cancellation of the
+     * subscription. To be precise, this method is called before
+     * {@link #onComplete()}, {@link #onError(Throwable) onError()}, or
+     * {@link #onCancel()}.
+     */
+    protected void onTermination() {
+        // Do nothing
+    }
+
+    private final class SubscriptionWrapper implements Subscription {
         final Subscription subscription;
         SubscriptionWrapper(Subscription s) {
             this.subscription = Objects.requireNonNull(s);
@@ -92,6 +101,7 @@ public void cancel() {
                     subscription.cancel();
                 } finally {
                     if (markCancelled()) {
+                        onTermination();
                         onCancel();
                     }
                 }
@@ -284,6 +294,7 @@ protected void onCancel() {
      */
     public final void complete(Throwable t) {
         if (markCompleted()) {
+            onTermination();
             logComplete(t);
             tryUnregister();
             t  = withError = Utils.getCompletionCause(t);