diff --git a/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java b/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java index b29ce5306a..3adc7a30b2 100644 --- a/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java +++ b/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java @@ -109,6 +109,10 @@ public interface AsyncHttpClientConfig { /** * Return the maximum time an {@link AsyncHttpClient} waits until the response is completed. + *

+ * By default this bounds each attempt within an exchange rather than the exchange as a whole: a redirect, a + * retry and an auth replay each start it again, so a chain of n hops may run for n times this value. Set + * {@link #isUseAbsoluteRequestDeadline()} to bound the exchange instead. * * @return the maximum time an {@link AsyncHttpClient} waits until the response is completed. */ @@ -142,6 +146,30 @@ default boolean isUseEventLoopTimeouts() { return false; } + /** + * Whether {@link #getRequestTimeout()} is a deadline for the whole exchange rather than for each attempt + * within it. + *

+ * A redirect, a retry and an auth replay all continue the same exchange on the same response future, but + * each builds its own timeout state. Anchoring the deadline on that state gives every hop a fresh budget, + * which is why a five-redirect chain can legitimately take six times the configured timeout today. Enabling + * this anchors it on when the exchange was submitted instead, so a later hop gets whatever is left and the + * caller's total wait is bounded by the one value. + *

+ * Off by default because turning it on shortens exchanges that rely on the per-attempt behaviour. A caller + * working to an end-to-end budget wants it on; {@link Request#getUseAbsoluteRequestDeadline()} sets it for a + * single request. + *

+ * As with every option on this interface, the {@code org.asynchttpclient.useAbsoluteRequestDeadline} + * property is read by {@link DefaultAsyncHttpClientConfig.Builder}, not here: an implementation of this + * interface that does not override this method gets {@code false} whatever the property says. + * + * @return {@code true} to treat the request timeout as a deadline for the whole exchange + */ + default boolean isUseAbsoluteRequestDeadline() { + return false; + } + /** * Is HTTP redirect enabled * diff --git a/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClientConfig.java b/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClientConfig.java index a1eed3cc97..7095a4cbca 100644 --- a/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClientConfig.java +++ b/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClientConfig.java @@ -63,6 +63,7 @@ import static org.asynchttpclient.config.AsyncHttpClientConfigDefaults.defaultEnabledProtocols; import static org.asynchttpclient.config.AsyncHttpClientConfigDefaults.defaultExpiredCookieEvictionDelay; import static org.asynchttpclient.config.AsyncHttpClientConfigDefaults.defaultFailedIpCooldownEnabled; +import static org.asynchttpclient.config.AsyncHttpClientConfigDefaults.defaultUseAbsoluteRequestDeadline; import static org.asynchttpclient.config.AsyncHttpClientConfigDefaults.defaultFailedIpCooldownPeriod; import static org.asynchttpclient.config.AsyncHttpClientConfigDefaults.defaultFilterInsecureCipherSuites; import static org.asynchttpclient.config.AsyncHttpClientConfigDefaults.defaultFollowRedirect; @@ -139,6 +140,7 @@ public class DefaultAsyncHttpClientConfig implements AsyncHttpClientConfig { private final int maxRequestRetry; private final LoadBalance loadBalance; private final boolean failedIpCooldownEnabled; + private final boolean useAbsoluteRequestDeadline; private final Duration failedIpCooldownPeriod; private final boolean disableUrlEncodingForBoundRequests; private final boolean useLaxCookieEncoder; @@ -245,6 +247,7 @@ private DefaultAsyncHttpClientConfig(// http int maxRequestRetry, LoadBalance loadBalance, boolean failedIpCooldownEnabled, + boolean useAbsoluteRequestDeadline, Duration failedIpCooldownPeriod, boolean disableUrlEncodingForBoundRequests, boolean useLaxCookieEncoder, @@ -351,6 +354,7 @@ private DefaultAsyncHttpClientConfig(// http this.maxRequestRetry = maxRequestRetry; this.loadBalance = loadBalance; this.failedIpCooldownEnabled = failedIpCooldownEnabled; + this.useAbsoluteRequestDeadline = useAbsoluteRequestDeadline; this.failedIpCooldownPeriod = failedIpCooldownPeriod; this.disableUrlEncodingForBoundRequests = disableUrlEncodingForBoundRequests; this.useLaxCookieEncoder = useLaxCookieEncoder; @@ -522,6 +526,11 @@ public boolean isFailedIpCooldownEnabled() { return failedIpCooldownEnabled; } + @Override + public boolean isUseAbsoluteRequestDeadline() { + return useAbsoluteRequestDeadline; + } + @Override public Duration getFailedIpCooldownPeriod() { return failedIpCooldownPeriod; @@ -946,6 +955,7 @@ public static class Builder { private int maxRequestRetry = defaultMaxRequestRetry(); private LoadBalance loadBalance = defaultLoadBalance(); private boolean failedIpCooldownEnabled = defaultFailedIpCooldownEnabled(); + private boolean useAbsoluteRequestDeadline = defaultUseAbsoluteRequestDeadline(); private Duration failedIpCooldownPeriod = defaultFailedIpCooldownPeriod(); private boolean disableUrlEncodingForBoundRequests = defaultDisableUrlEncodingForBoundRequests(); private boolean useLaxCookieEncoder = defaultUseLaxCookieEncoder(); @@ -1055,6 +1065,7 @@ public Builder(AsyncHttpClientConfig config) { maxRequestRetry = config.getMaxRequestRetry(); loadBalance = config.getLoadBalance(); failedIpCooldownEnabled = config.isFailedIpCooldownEnabled(); + useAbsoluteRequestDeadline = config.isUseAbsoluteRequestDeadline(); failedIpCooldownPeriod = config.getFailedIpCooldownPeriod(); disableUrlEncodingForBoundRequests = config.isDisableUrlEncodingForBoundRequests(); useLaxCookieEncoder = config.isUseLaxCookieEncoder(); @@ -1255,6 +1266,17 @@ public Builder setFailedIpCooldownEnabled(boolean failedIpCooldownEnabled) { return this; } + /** + * @param useAbsoluteRequestDeadline whether the request timeout is a deadline for the whole exchange + * rather than for each attempt within it; see + * {@link AsyncHttpClientConfig#isUseAbsoluteRequestDeadline()} + * @return this + */ + public Builder setUseAbsoluteRequestDeadline(boolean useAbsoluteRequestDeadline) { + this.useAbsoluteRequestDeadline = useAbsoluteRequestDeadline; + return this; + } + /** * @param failedIpCooldownPeriod how long a failed IP is deprioritized before it is re-probed; * {@code null} resets to the default. Must not be negative; use @@ -1773,6 +1795,7 @@ public DefaultAsyncHttpClientConfig build() { maxRequestRetry, loadBalance, failedIpCooldownEnabled, + useAbsoluteRequestDeadline, failedIpCooldownPeriod, disableUrlEncodingForBoundRequests, useLaxCookieEncoder, diff --git a/client/src/main/java/org/asynchttpclient/DefaultRequest.java b/client/src/main/java/org/asynchttpclient/DefaultRequest.java index c8e44e338f..3a4885601f 100644 --- a/client/src/main/java/org/asynchttpclient/DefaultRequest.java +++ b/client/src/main/java/org/asynchttpclient/DefaultRequest.java @@ -63,6 +63,7 @@ public class DefaultRequest implements Request { private final @Nullable Realm realm; private final @Nullable File file; private final @Nullable Boolean followRedirect; + private final @Nullable Boolean useAbsoluteRequestDeadline; private final Duration requestTimeout; private final Duration readTimeout; private final long rangeOffset; @@ -99,6 +100,48 @@ public DefaultRequest(String method, @Nullable Charset charset, ChannelPoolPartitioning channelPoolPartitioning, NameResolver nameResolver) { + this(method, uri, address, localAddress, headers, cookies, byteData, compositeByteData, stringData, + byteBufferData, byteBufData, streamData, bodyGenerator, formParams, bodyParts, virtualHost, + proxyServer, realm, file, followRedirect, requestTimeout, readTimeout, rangeOffset, charset, + channelPoolPartitioning, nameResolver, null); + } + + /** + * The full set of fields a request carries, called only by {@link RequestBuilderBase#build()}. Not public and + * not part of the API: the constructor above is what outside callers compile against, so this one is free to + * take another field without pinning a signature or asking the next reader to keep two parameter lists of + * reference types in step by eye. + * + * @param useAbsoluteRequestDeadline whether {@code requestTimeout} bounds the whole exchange rather than + * each attempt within it, or null to defer to the client config + */ + DefaultRequest(String method, + Uri uri, + @Nullable InetAddress address, + @Nullable InetAddress localAddress, + HttpHeaders headers, + List cookies, + byte @Nullable [] byteData, + @Nullable List compositeByteData, + @Nullable String stringData, + @Nullable ByteBuffer byteBufferData, + @Nullable ByteBuf byteBufData, + @Nullable InputStream streamData, + @Nullable BodyGenerator bodyGenerator, + List formParams, + List bodyParts, + @Nullable String virtualHost, + @Nullable ProxyServer proxyServer, + @Nullable Realm realm, + @Nullable File file, + @Nullable Boolean followRedirect, + @Nullable Duration requestTimeout, + @Nullable Duration readTimeout, + long rangeOffset, + @Nullable Charset charset, + ChannelPoolPartitioning channelPoolPartitioning, + NameResolver nameResolver, + @Nullable Boolean useAbsoluteRequestDeadline) { this.method = method; this.uri = uri; this.address = address; @@ -119,6 +162,7 @@ public DefaultRequest(String method, this.realm = realm; this.file = file; this.followRedirect = followRedirect; + this.useAbsoluteRequestDeadline = useAbsoluteRequestDeadline; this.requestTimeout = requestTimeout == null ? Duration.ZERO : requestTimeout; this.readTimeout = readTimeout == null ? Duration.ZERO : readTimeout; this.rangeOffset = rangeOffset; @@ -232,6 +276,11 @@ public List getBodyParts() { return followRedirect; } + @Override + public @Nullable Boolean getUseAbsoluteRequestDeadline() { + return useAbsoluteRequestDeadline; + } + @Override public Duration getRequestTimeout() { return requestTimeout; diff --git a/client/src/main/java/org/asynchttpclient/Request.java b/client/src/main/java/org/asynchttpclient/Request.java index 1d95016b36..1ec85fd62b 100644 --- a/client/src/main/java/org/asynchttpclient/Request.java +++ b/client/src/main/java/org/asynchttpclient/Request.java @@ -172,6 +172,17 @@ public interface Request { @Nullable Boolean getFollowRedirect(); + /** + * Whether {@link #getRequestTimeout()} is a deadline for the whole exchange rather than for each attempt + * within it. See {@link AsyncHttpClientConfig#isUseAbsoluteRequestDeadline()}. + * + * @return the override, or null to use the config value + */ + @Nullable + default Boolean getUseAbsoluteRequestDeadline() { + return null; + } + /** * @return the request timeout. Non zero values means "override config value". */ diff --git a/client/src/main/java/org/asynchttpclient/RequestBuilderBase.java b/client/src/main/java/org/asynchttpclient/RequestBuilderBase.java index b185cbdc90..42a5b28b2c 100644 --- a/client/src/main/java/org/asynchttpclient/RequestBuilderBase.java +++ b/client/src/main/java/org/asynchttpclient/RequestBuilderBase.java @@ -87,6 +87,7 @@ public abstract class RequestBuilderBase> { protected @Nullable Realm realm; protected @Nullable File file; protected @Nullable Boolean followRedirect; + protected @Nullable Boolean useAbsoluteRequestDeadline; protected @Nullable Duration requestTimeout; protected @Nullable Duration readTimeout; protected long rangeOffset; @@ -165,6 +166,7 @@ protected RequestBuilderBase(Request prototype, boolean disableUrlEncoding, bool realm = prototype.getRealm(); file = prototype.getFile(); followRedirect = prototype.getFollowRedirect(); + useAbsoluteRequestDeadline = prototype.getUseAbsoluteRequestDeadline(); requestTimeout = prototype.getRequestTimeout(); readTimeout = prototype.getReadTimeout(); rangeOffset = prototype.getRangeOffset(); @@ -598,6 +600,17 @@ public T setRealm(Realm realm) { return asDerivedType(); } + /** + * @param useAbsoluteRequestDeadline whether this request's timeout is a deadline for the whole exchange + * rather than for each attempt within it, overriding + * {@link AsyncHttpClientConfig#isUseAbsoluteRequestDeadline()} + * @return {@code this} + */ + public T setUseAbsoluteRequestDeadline(boolean useAbsoluteRequestDeadline) { + this.useAbsoluteRequestDeadline = useAbsoluteRequestDeadline; + return asDerivedType(); + } + public T setFollowRedirect(boolean followRedirect) { this.followRedirect = followRedirect; return asDerivedType(); @@ -685,7 +698,9 @@ private RequestBuilderBase executeSignatureCalculator() { rb.realm = realm; rb.file = file; rb.followRedirect = followRedirect; + rb.useAbsoluteRequestDeadline = useAbsoluteRequestDeadline; rb.requestTimeout = requestTimeout; + rb.readTimeout = readTimeout; rb.rangeOffset = rangeOffset; rb.charset = charset; rb.channelPoolPartitioning = channelPoolPartitioning; @@ -755,6 +770,7 @@ public Request build() { rb.rangeOffset, rb.charset, rb.channelPoolPartitioning, - rb.nameResolver); + rb.nameResolver, + rb.useAbsoluteRequestDeadline); } } diff --git a/client/src/main/java/org/asynchttpclient/config/AsyncHttpClientConfigDefaults.java b/client/src/main/java/org/asynchttpclient/config/AsyncHttpClientConfigDefaults.java index 50fcd723aa..25778dee98 100644 --- a/client/src/main/java/org/asynchttpclient/config/AsyncHttpClientConfigDefaults.java +++ b/client/src/main/java/org/asynchttpclient/config/AsyncHttpClientConfigDefaults.java @@ -63,6 +63,7 @@ public final class AsyncHttpClientConfigDefaults { public static final String MAX_REQUEST_RETRY_CONFIG = "maxRequestRetry"; public static final String LOAD_BALANCE_CONFIG = "loadBalance"; public static final String FAILED_IP_COOLDOWN_ENABLED_CONFIG = "failedIpCooldownEnabled"; + public static final String USE_ABSOLUTE_REQUEST_DEADLINE_CONFIG = "useAbsoluteRequestDeadline"; public static final String FAILED_IP_COOLDOWN_PERIOD_CONFIG = "failedIpCooldownPeriod"; public static final String DISABLE_URL_ENCODING_FOR_BOUND_REQUESTS_CONFIG = "disableUrlEncodingForBoundRequests"; public static final String USE_LAX_COOKIE_ENCODER_CONFIG = "useLaxCookieEncoder"; @@ -188,6 +189,10 @@ public static boolean defaultFailedIpCooldownEnabled() { return AsyncHttpClientConfigHelper.getAsyncHttpClientConfig().getBoolean(ASYNC_CLIENT_CONFIG_ROOT + FAILED_IP_COOLDOWN_ENABLED_CONFIG); } + public static boolean defaultUseAbsoluteRequestDeadline() { + return AsyncHttpClientConfigHelper.getAsyncHttpClientConfig().getBoolean(ASYNC_CLIENT_CONFIG_ROOT + USE_ABSOLUTE_REQUEST_DEADLINE_CONFIG); + } + public static Duration defaultFailedIpCooldownPeriod() { return AsyncHttpClientConfigHelper.getAsyncHttpClientConfig().getDuration(ASYNC_CLIENT_CONFIG_ROOT + FAILED_IP_COOLDOWN_PERIOD_CONFIG); } diff --git a/client/src/main/java/org/asynchttpclient/netty/NettyResponseFuture.java b/client/src/main/java/org/asynchttpclient/netty/NettyResponseFuture.java index bcd9032833..c3616e5bda 100755 --- a/client/src/main/java/org/asynchttpclient/netty/NettyResponseFuture.java +++ b/client/src/main/java/org/asynchttpclient/netty/NettyResponseFuture.java @@ -92,6 +92,10 @@ public final class NettyResponseFuture implements ListenableFuture { .newUpdater(NettyResponseFuture.class, Object.class, "partitionKeyLock"); private final long start = unpreciseMillisTime(); + // Wall clock is what start reports, and a deadline anchored on it for the length of a whole exchange would + // be moved by any clock correction that lands mid-chain: a step back would hand a later hop a budget it + // never had, a step forward would abort it on a healthy connection. Elapsed time is measured from here. + private final long startNanos = System.nanoTime(); private final ChannelPoolPartitioning connectionPoolPartitioning; private final ConnectionSemaphore connectionSemaphore; // Not final: a filter replay can retarget this future at a different origin, reached through a @@ -154,6 +158,9 @@ public final class NettyResponseFuture implements ListenableFuture { // future no longer takes, which is how a connection through a proxy comes to be offered as a direct one. // Volatile: the mutators run on the redirect and replay paths while reads happen on other threads. private volatile Object basePartitionKeyCache; + // Read when a TimeoutsHolder is built, which happens on the caller thread, an event loop or the timer + // thread depending on the path, so it is published rather than plain. + private volatile boolean useAbsoluteRequestDeadline; public NettyResponseFuture(Request originalRequest, AsyncHandler asyncHandler, @@ -598,6 +605,14 @@ public long getStart() { return start; } + /** + * When this exchange was submitted, on a monotonic clock, for measuring how much of a deadline spanning the + * whole exchange it has spent. Comparable only with other {@link System#nanoTime()} readings. + */ + public long getStartNanos() { + return startNanos; + } + public Object getPartitionKey() { Object override = partitionKeyOverride; if (override != null) { @@ -731,6 +746,21 @@ public void acquirePartitionLockLazily(boolean nonBlocking) throws IOException { } } + /** + * Whether this exchange's request timeout is a deadline for the exchange as a whole. Resolved once, from the + * request the caller submitted and the client config, and then kept here rather than re-read per hop: a + * redirect rebuilds the request from a hand-picked set of fields, so anything carried only on the request + * would silently revert to the config value partway through the exchange, which is exactly the case this + * setting exists for. + */ + public boolean isUseAbsoluteRequestDeadline() { + return useAbsoluteRequestDeadline; + } + + public void setUseAbsoluteRequestDeadline(boolean useAbsoluteRequestDeadline) { + this.useAbsoluteRequestDeadline = useAbsoluteRequestDeadline; + } + public Realm getRealm() { return realm; } diff --git a/client/src/main/java/org/asynchttpclient/netty/handler/intercept/Redirect30xInterceptor.java b/client/src/main/java/org/asynchttpclient/netty/handler/intercept/Redirect30xInterceptor.java index b1c70c257d..80cac6d4f9 100644 --- a/client/src/main/java/org/asynchttpclient/netty/handler/intercept/Redirect30xInterceptor.java +++ b/client/src/main/java/org/asynchttpclient/netty/handler/intercept/Redirect30xInterceptor.java @@ -139,7 +139,17 @@ public boolean exitAfterHandlingRedirect(Channel channel, NettyResponseFuture .setNameResolver(request.getNameResolver()) .setProxyServer(request.getProxyServer()) .setRealm(stripAuth ? null : request.getRealm()) - .setRequestTimeout(request.getRequestTimeout()); + .setRequestTimeout(request.getRequestTimeout()) + .setReadTimeout(request.getReadTimeout()); + + // The exchange holds the deadline flag on its future, so a hop that does not carry it forward + // leaves the request saying something the exchange is not doing, which is what a filter or a + // signature calculator reads. Set only when the request has one: the setter takes a primitive, + // and null is how a request defers to the client config. + Boolean useAbsoluteRequestDeadline = request.getUseAbsoluteRequestDeadline(); + if (useAbsoluteRequestDeadline != null) { + requestBuilder.setUseAbsoluteRequestDeadline(useAbsoluteRequestDeadline); + } if (stripAuth) { future.setRealm(null); diff --git a/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java b/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java index 47775c4479..c142bc62ab 100755 --- a/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java +++ b/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java @@ -81,6 +81,7 @@ import org.asynchttpclient.proxy.ProxyType; import org.asynchttpclient.resolver.RequestHostnameResolver; import org.asynchttpclient.uri.Uri; +import org.asynchttpclient.util.StringBuilderPool; import org.asynchttpclient.ws.WebSocketUpgradeHandler; import org.jetbrains.annotations.Nullable; @@ -98,9 +99,11 @@ import java.util.Locale; import java.util.Map; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Consumer; +import static org.asynchttpclient.util.DateUtils.unpreciseMillisTime; import static io.netty.handler.codec.http.HttpHeaderNames.EXPECT; import static java.util.Collections.singletonList; import static java.util.Objects.requireNonNull; @@ -111,6 +114,7 @@ import static org.asynchttpclient.util.HttpUtils.GZIP_DEFLATE; import static org.asynchttpclient.util.HttpUtils.GZIP_DEFLATE_HPACK; import static org.asynchttpclient.util.HttpUtils.hostHeader; +import static org.asynchttpclient.util.HttpUtils.useAbsoluteRequestDeadline; import static org.asynchttpclient.util.MiscUtils.getCause; import static org.asynchttpclient.util.ProxyUtils.getProxyServer; @@ -408,9 +412,10 @@ private ListenableFuture sendRequestWithOpenChannel(NettyResponseFuture ListenableFuture sendRequestWithNewChannel(Request request, Proxy abort(null, future, new UnknownHostException("No addresses resolved for " + request.getUri().getHost())); return future; } - scheduleRequestTimeout(future, roundRobinAddresses.get(0)); + if (!scheduleRequestTimeout(future, roundRobinAddresses.get(0))) { + return future; + } connectWithAddresses(request, proxy, future, asyncHandler, roundRobinAddresses); return future; } @@ -591,16 +598,16 @@ private Future> resolveAddresses(Request request, Pr if (proxy != null && !proxy.isIgnoredForHost(uri.getHost()) && proxy.getProxyType().isHttp()) { int port = ProxyType.HTTPS.equals(proxy.getProxyType()) || uri.isSecured() ? proxy.getSecuredPort() : proxy.getPort(); InetSocketAddress unresolvedRemoteAddress = InetSocketAddress.createUnresolved(proxy.getHost(), port); - if (scheduleTimeout) { - scheduleRequestTimeout(future, unresolvedRemoteAddress); + if (scheduleTimeout && !scheduleRequestTimeout(future, unresolvedRemoteAddress)) { + return abortedResolution(future); } return resolveHostname(request, unresolvedRemoteAddress, asyncHandler); } else { int port = uri.getExplicitPort(); InetSocketAddress unresolvedRemoteAddress = InetSocketAddress.createUnresolved(uri.getHost(), port); - if (scheduleTimeout) { - scheduleRequestTimeout(future, unresolvedRemoteAddress); + if (scheduleTimeout && !scheduleRequestTimeout(future, unresolvedRemoteAddress)) { + return abortedResolution(future); } if (request.getAddress() != null) { @@ -632,6 +639,8 @@ private NettyResponseFuture newNettyResponseFuture(Request request, Async connectionSemaphore, proxyServer); + future.setUseAbsoluteRequestDeadline(useAbsoluteRequestDeadline(config, request)); + String expectHeader = request.getHeaders().get(EXPECT); if (HttpHeaderValues.CONTINUE.contentEqualsIgnoreCase(expectHeader)) { future.setDontWriteBodyBecauseExpectContinue(true); @@ -1081,26 +1090,40 @@ private static void configureTransferAdapter(AsyncHandler handler, HttpReques ((TransferCompletionHandler) handler).headers(h); } - private void scheduleRequestTimeout(NettyResponseFuture nettyResponseFuture, - InetSocketAddress originalRemoteAddress) { - scheduleRequestTimeout(nettyResponseFuture, originalRemoteAddress, null); + private boolean scheduleRequestTimeout(NettyResponseFuture nettyResponseFuture, + InetSocketAddress originalRemoteAddress) { + return scheduleRequestTimeout(nettyResponseFuture, originalRemoteAddress, null); } /** + * Arms the timeouts for the attempt about to be made, unless the exchange has no time left to make it in. + * Every attempt passes through here, whether it is the first or a redirect, an auth replay or a retry, and + * it is the last point before the request is written -- so it is where a deadline is worth one more look. + * Arming at zero instead would abort the attempt, but only after a connection permit had been taken, a + * connection taken and the request written: a 307 would put its body on the redirect target and then hand + * the caller a TimeoutException that reads as though nothing had been sent. + * * @param channel the channel the exchange will run on when it is already known, so the timeout can be armed * on the loop that owns it. Null on the connect path: the timeout is armed before the channel * exists, deliberately, so that it also bounds address resolution and the connect itself, and * {@code TimeoutsHolder#rehomeOn} moves it onto the loop once there is one. + * @return whether the attempt may go ahead. When {@code false} the exchange has already been aborted. */ - private void scheduleRequestTimeout(NettyResponseFuture nettyResponseFuture, - InetSocketAddress originalRemoteAddress, - @Nullable Channel channel) { + private boolean scheduleRequestTimeout(NettyResponseFuture nettyResponseFuture, + InetSocketAddress originalRemoteAddress, + @Nullable Channel channel) { + if (TimeoutsHolder.remainingBudget(config, nettyResponseFuture) <= 0L) { + abort(nettyResponseFuture.channel(), nettyResponseFuture, + new TimeoutException(deadlinePassedMessage(nettyResponseFuture.getTargetRequest(), nettyResponseFuture))); + return false; + } nettyResponseFuture.touch(); TimeoutsHolder timeoutsHolder = new TimeoutsHolder(nettyTimer, timeoutExecutor(channel), nettyResponseFuture, this, config, originalRemoteAddress); // Arms the timeout as a part of installing the holder, which is why the pooled path attaches the // channel first: an expiry that lands immediately reaches the channel only through the future. nettyResponseFuture.setTimeoutsHolder(timeoutsHolder); + return true; } /** @@ -1213,6 +1236,24 @@ public void sendNextRequest(final Request request, final NettyResponseFuture sendRequest(request, future.getAsyncHandler(), future); } + /** + * A resolution that will not be attempted, for an attempt the exchange has no time left to make. The + * exchange is aborted before this is returned, so the failure carried here only stops the listener from + * carrying on with a connect. + */ + private static Future> abortedResolution(NettyResponseFuture future) { + return ImmediateEventExecutor.INSTANCE.newFailedFuture( + new TimeoutException(deadlinePassedMessage(future.getTargetRequest(), future))); + } + + private static String deadlinePassedMessage(Request request, NettyResponseFuture future) { + return StringBuilderPool.DEFAULT.stringBuilder() + .append("Request timeout to ").append(request.getUri().getHost()) + .append(':').append(request.getUri().getExplicitPort()) + .append(" after ").append(unpreciseMillisTime() - future.getStart()) + .append(" ms, before the request was sent").toString(); + } + private static void validateWebSocketRequest(Request request, AsyncHandler asyncHandler) { Uri uri = request.getUri(); boolean isWs = uri.isWebSocket(); diff --git a/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java b/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java index ba76939958..b1af44b301 100755 --- a/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java +++ b/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java @@ -48,6 +48,7 @@ public class TimeoutsHolder { private final NettyRequestSender requestSender; private final long requestTimeoutMillisTime; private final long requestTimeoutValue; + private final boolean absoluteDeadline; private final long readTimeoutValue; private final boolean useEventLoopTimeouts; private final @Nullable RequestTimeoutTimerTask requestTimeoutTask; @@ -81,14 +82,18 @@ public TimeoutsHolder(Timer nettyTimer, @Nullable EventExecutor eventExecutor, N final long readTimeoutInMs = targetRequest.getReadTimeout().toMillis(); readTimeoutValue = readTimeoutInMs == 0 ? config.getReadTimeout().toMillis() : readTimeoutInMs; - long requestTimeoutInMs = targetRequest.getRequestTimeout().toMillis(); - if (requestTimeoutInMs == 0) { - requestTimeoutInMs = config.getRequestTimeout().toMillis(); - } + long requestTimeoutInMs = requestTimeout(config, targetRequest); requestTimeoutValue = requestTimeoutInMs; + absoluteDeadline = nettyResponseFuture.isUseAbsoluteRequestDeadline(); if (requestTimeoutInMs > -1) { - requestTimeoutMillisTime = unpreciseMillisTime() + requestTimeoutInMs; + // A redirect, a retry or an auth replay builds a new holder for the same future. Giving each of + // those hops the configured timeout lets a chain of n hops run for n times it; netting off what the + // exchange has already spent bounds it as a whole instead. Which one applies is the caller's + // choice, per request or per client. Left negative when the deadline is already behind us, which is + // what stops startReadTimeout arming a sibling for an exchange that is over. + requestTimeoutMillisTime = unpreciseMillisTime() + + (absoluteDeadline ? remainingBudget(requestTimeoutInMs, nettyResponseFuture) : requestTimeoutInMs); requestTimeoutTask = new RequestTimeoutTimerTask(nettyResponseFuture, requestSender, this, requestTimeoutInMs); } else { requestTimeoutMillisTime = -1L; @@ -108,12 +113,55 @@ public TimeoutsHolder(Timer nettyTimer, @Nullable EventExecutor eventExecutor, N */ public void start() { if (requestTimeoutTask != null) { - // The configured duration rather than the remaining time: this runs within microseconds of the - // constructor, and reading the clock again would only expose the deadline to a step between the two. - arm(requestTimeoutTask, requestTimeoutValue); + // Per attempt, the configured duration: this runs within microseconds of the constructor, so + // reading the clock again would only expose the deadline to a step between the two reads. An + // absolute deadline was anchored before this holder existed, so there the remainder is the budget, + // floored at zero: a task armed at zero still runs, and running is how the exchange gets failed. + arm(requestTimeoutTask, absoluteDeadline + ? Math.max(remainingBudget(requestTimeoutValue, nettyResponseFuture), 0L) : requestTimeoutValue); } } + /** + * How much of a deadline spanning the whole exchange is left, in milliseconds, negative once it has passed. + * {@link Long#MAX_VALUE} when the timeout is per attempt or disabled, neither of which bounds an exchange as + * a whole: an attempt is then given the configured timeout of its own however long the exchange has run. + *

+ * Measured from the future's monotonic start rather than by comparing wall clocks across hops, so a clock + * correction landing mid-chain cannot move the deadline. Static, and asked of the future rather than of a + * holder, because a caller deciding whether a request is still worth sending has the future in hand before + * any holder exists for the attempt it is about to make. + * + * @see org.asynchttpclient.AsyncHttpClientConfig#isUseAbsoluteRequestDeadline() + */ + public static long remainingBudget(AsyncHttpClientConfig config, NettyResponseFuture nettyResponseFuture) { + if (!nettyResponseFuture.isUseAbsoluteRequestDeadline()) { + return Long.MAX_VALUE; + } + return remainingBudget(requestTimeout(config, nettyResponseFuture.getTargetRequest()), nettyResponseFuture); + } + + private static long remainingBudget(long requestTimeoutInMs, NettyResponseFuture nettyResponseFuture) { + if (requestTimeoutInMs <= -1) { + return Long.MAX_VALUE; + } + long spent = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - nettyResponseFuture.getStartNanos()); + return requestTimeoutInMs - spent; + } + + /** + * The request timeout in force for {@code request}: its own, or the client's when it does not carry one. + */ + private static long requestTimeout(AsyncHttpClientConfig config, Request request) { + long requestTimeoutInMs = request.getRequestTimeout().toMillis(); + return requestTimeoutInMs == 0 ? config.getRequestTimeout().toMillis() : requestTimeoutInMs; + } + + // Visible for testing: the instant this holder's request timeout is due, as a wall-clock reading. + long requestTimeoutMillisTime() { + return requestTimeoutMillisTime; + } + /** * Moves this exchange's timeouts onto {@code executor}, the loop of the channel it turned out to run on. The * connect path arms the request timeout before there is a channel -- deliberately, since it bounds address @@ -178,8 +226,8 @@ private static void release(@Nullable TimeoutTimerTask task) { } private long remainingRequestTimeout() { - // A deadline already behind us is armed at zero rather than negative, so the task still runs and still - // cancels its read-timeout sibling, which is bookkeeping only it does. + // Floored at zero rather than passed on negative: a scheduler has no use for a negative delay, and the + // task has to run either way, since running is what fails the exchange. return Math.max(requestTimeoutMillisTime - unpreciseMillisTime(), 0L); } diff --git a/client/src/main/java/org/asynchttpclient/util/HttpUtils.java b/client/src/main/java/org/asynchttpclient/util/HttpUtils.java index 4e8d802575..2b970595be 100644 --- a/client/src/main/java/org/asynchttpclient/util/HttpUtils.java +++ b/client/src/main/java/org/asynchttpclient/util/HttpUtils.java @@ -142,6 +142,11 @@ public static boolean followRedirect(AsyncHttpClientConfig config, Request reque return request.getFollowRedirect() != null ? request.getFollowRedirect() : config.isFollowRedirect(); } + public static boolean useAbsoluteRequestDeadline(AsyncHttpClientConfig config, Request request) { + Boolean override = request.getUseAbsoluteRequestDeadline(); + return override != null ? override : config.isUseAbsoluteRequestDeadline(); + } + public static ByteBuffer urlEncodeFormParams(List params, Charset charset) { return StringUtils.charSequence2ByteBuffer(urlEncodeFormParams0(params, charset), US_ASCII); } diff --git a/client/src/main/resources/org/asynchttpclient/config/ahc-default.properties b/client/src/main/resources/org/asynchttpclient/config/ahc-default.properties index 34fb663803..1179088cab 100644 --- a/client/src/main/resources/org/asynchttpclient/config/ahc-default.properties +++ b/client/src/main/resources/org/asynchttpclient/config/ahc-default.properties @@ -27,6 +27,7 @@ org.asynchttpclient.keepAlive=true org.asynchttpclient.maxRequestRetry=5 org.asynchttpclient.loadBalance=DEFAULT org.asynchttpclient.failedIpCooldownEnabled=true +org.asynchttpclient.useAbsoluteRequestDeadline=false org.asynchttpclient.failedIpCooldownPeriod=PT10S org.asynchttpclient.disableUrlEncodingForBoundRequests=false org.asynchttpclient.useLaxCookieEncoder=false diff --git a/client/src/test/java/org/asynchttpclient/AbsoluteRequestDeadlineTest.java b/client/src/test/java/org/asynchttpclient/AbsoluteRequestDeadlineTest.java new file mode 100644 index 0000000000..b292c76247 --- /dev/null +++ b/client/src/test/java/org/asynchttpclient/AbsoluteRequestDeadlineTest.java @@ -0,0 +1,248 @@ +/* + * Copyright (c) 2026 AsyncHttpClient Project. All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.asynchttpclient; + +import io.github.artsok.RepeatedIfExceptionsTest; +import io.netty.handler.codec.http.HttpHeaderNames; +import io.netty.util.HashedWheelTimer; +import io.netty.util.concurrent.DefaultThreadFactory; +import org.asynchttpclient.testserver.HttpServer; +import org.asynchttpclient.testserver.HttpTest; +import org.jetbrains.annotations.Nullable; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; + +import java.io.IOException; +import java.time.Duration; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; + +import static org.asynchttpclient.Dsl.config; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * {@link AsyncHttpClientConfig#isUseAbsoluteRequestDeadline()} decides whether the request timeout bounds the + * whole exchange or each attempt within it. A redirect builds a fresh {@code TimeoutsHolder} for the same + * future, so with the deadline anchored on the holder each hop gets a budget of its own, and with it anchored + * on the future a later hop gets only what is left. + *

+ * Timing-based, so repeated: the margins are wide (a 600 ms budget against hops of 400 ms) but a loaded CI box + * can still miss one. + */ +public class AbsoluteRequestDeadlineTest extends HttpTest { + + private static final Duration BUDGET = Duration.ofMillis(600); + private static final long HOP_DELAY_MS = 400; + private static final String FIRST_HOP = "/foo/bar"; + private static final String SECOND_HOP = "/foo/bar2"; + + private HttpServer server; + // Coarse on purpose, for the one case that needs the request timeout not to fire: a wheel answers a + // deadline on its first tick at or after it, so at this granularity nothing expires inside a test. + private HashedWheelTimer stalledTimer; + + @BeforeEach + public void start() throws Throwable { + server = new HttpServer(); + server.start(); + stalledTimer = new HashedWheelTimer(new DefaultThreadFactory("ahc-stalled-timer", true), + 30, TimeUnit.SECONDS, 512, false); + } + + @AfterEach + public void stop() throws Throwable { + server.close(); + stalledTimer.stop(); + } + + @RepeatedIfExceptionsTest(repeats = 5) + public void byDefaultEachHopGetsItsOwnBudget() throws Throwable { + // Two hops of 400 ms against a 600 ms budget. Each hop on its own fits, the pair does not, so with a + // per-attempt timeout the exchange completes. + enqueueTwoDelayedHops(); + + Outcome outcome = runAndAwait(baseConfig(), null); + + outcome.assertReachedTheSecondHop(); + } + + @RepeatedIfExceptionsTest(repeats = 5) + public void withAnAbsoluteDeadlineTheChainCannotOutrunTheBudget() throws Throwable { + enqueueTwoDelayedHops(); + + Outcome outcome = runAndAwait(baseConfig().setUseAbsoluteRequestDeadline(true), null); + + outcome.assertTimedOut(); + } + + @RepeatedIfExceptionsTest(repeats = 5) + public void aRequestCanAskForAnAbsoluteDeadlineOnAPerAttemptClient() throws Throwable { + enqueueTwoDelayedHops(); + + Outcome outcome = runAndAwait(baseConfig(), Boolean.TRUE); + + outcome.assertTimedOut(); + } + + @RepeatedIfExceptionsTest(repeats = 5) + public void aRequestCanOptOutOfAnAbsoluteDeadlineClient() throws Throwable { + enqueueTwoDelayedHops(); + + Outcome outcome = runAndAwait(baseConfig().setUseAbsoluteRequestDeadline(true), Boolean.FALSE); + + outcome.assertReachedTheSecondHop(); + } + + @RepeatedIfExceptionsTest(repeats = 5) + public void aSingleHopStillGetsTheWholeBudget() throws Throwable { + // Guards the other direction: with a deadline, the first hop must not be handed a shortened budget. + enqueueDelayed(HOP_DELAY_MS, 200, null); + + Outcome outcome = runAndAwait(baseConfig().setUseAbsoluteRequestDeadline(true), null); + + outcome.assertCompletedAt(FIRST_HOP); + } + + @RepeatedIfExceptionsTest(repeats = 5) + public void aHopWithNothingLeftToSpendIsNeverSent() throws Throwable { + // The first hop answers after the budget is gone, and the timer is too coarse to have expired the + // exchange in the meantime. That is the window in which a redirect used to be written anyway: a permit + // taken, a connection taken, the body on the wire, and only then a TimeoutException that reads to the + // caller as though nothing had been sent. + AtomicBoolean secondHopServed = new AtomicBoolean(); + enqueueDelayed(BUDGET.toMillis() + HOP_DELAY_MS, 302, SECOND_HOP); + server.enqueueResponse(response -> { + secondHopServed.set(true); + response.setStatus(200); + }); + + Outcome outcome = runAndAwait(baseConfig() + .setNettyTimer(stalledTimer) + .setUseAbsoluteRequestDeadline(true), null); + + outcome.assertTimedOutBeforeSending(); + assertFalse(secondHopServed.get(), "the redirect target was sent a request with no budget left"); + } + + private DefaultAsyncHttpClientConfig.Builder baseConfig() { + return config().setRequestTimeout(BUDGET).setFollowRedirect(true).setMaxRedirects(5); + } + + private void enqueueTwoDelayedHops() { + enqueueDelayed(HOP_DELAY_MS, 302, SECOND_HOP); + enqueueDelayed(HOP_DELAY_MS, 200, null); + } + + /** + * What the exchange ended as. The passing cases assert where it ended and not merely that nothing was + * thrown: a dropped {@code Location} header, or redirects turned off, would satisfy "no exception" having + * run one hop, which is the opposite of what they are for. + */ + private static final class Outcome { + + private final @Nullable Throwable cause; + private final @Nullable Response response; + + private Outcome(@Nullable Throwable cause, @Nullable Response response) { + this.cause = cause; + this.response = response; + } + + void assertCompletedAt(String path) { + assertNull(cause, "the exchange was not meant to fail, got " + cause); + assertNotNull(response, "the exchange neither failed nor produced a response"); + assertEquals(200, response.getStatusCode(), "expected the final 200"); + assertEquals(path, response.getUri().getPath(), "the exchange ended on the wrong hop"); + } + + void assertReachedTheSecondHop() { + assertCompletedAt(SECOND_HOP); + } + + void assertTimedOut() { + assertNotNull(cause, "the exchange should have run out of budget"); + assertEquals(TimeoutException.class, cause.getClass(), "expected a request timeout, got " + cause); + } + + /** + * That the exchange was failed by the check before the request was written, rather than by a timeout + * armed at zero expiring once it had been. The message is the only thing that tells the two apart. + */ + void assertTimedOutBeforeSending() { + assertTimedOut(); + assertTrue(cause.getMessage().contains("before the request was sent"), + "expected the deadline to be caught before the write, got " + cause.getMessage()); + } + } + + /** + * Answers after {@code delayMs}, so the hop consumes a known slice of the budget before the client sees a + * status at all. + */ + private void enqueueDelayed(long delayMs, int status, @Nullable String location) { + server.enqueueResponse(response -> { + try { + Thread.sleep(delayMs); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException(e); + } + response.setStatus(status); + if (location != null) { + response.setHeader(HttpHeaderNames.LOCATION.toString(), location); + } + }); + } + + private Outcome runAndAwait(DefaultAsyncHttpClientConfig.Builder builder, + @Nullable Boolean perRequestOverride) throws Throwable { + AtomicReference cause = new AtomicReference<>(); + AtomicReference completed = new AtomicReference<>(); + CountDownLatch settled = new CountDownLatch(1); + + withClient(builder).run(client -> withServer(server).run(server -> { + BoundRequestBuilder request = client.prepareGet(server.getHttpUrl() + FIRST_HOP); + if (perRequestOverride != null) { + request.setUseAbsoluteRequestDeadline(perRequestOverride); + } + request.execute(new AsyncCompletionHandler() { + @Override + public Void onCompleted(Response response) { + completed.set(response); + settled.countDown(); + return null; + } + + @Override + public void onThrowable(Throwable t) { + cause.set(t); + settled.countDown(); + } + }); + + assertTrue(settled.await(30, TimeUnit.SECONDS), "the exchange neither completed nor failed"); + })); + + return new Outcome(cause.get(), completed.get()); + } +} diff --git a/client/src/test/java/org/asynchttpclient/AsyncHttpClientDefaultsTest.java b/client/src/test/java/org/asynchttpclient/AsyncHttpClientDefaultsTest.java index d125a9fa48..39c7bab8f7 100644 --- a/client/src/test/java/org/asynchttpclient/AsyncHttpClientDefaultsTest.java +++ b/client/src/test/java/org/asynchttpclient/AsyncHttpClientDefaultsTest.java @@ -78,6 +78,18 @@ public void testDefaultConnectionTtl() { testDurationSystemProperty("connectionTtl", "defaultConnectionTtl", "PT0.1S"); } + @RepeatedIfExceptionsTest(repeats = 5) + public void testDefaultUseAbsoluteRequestDeadline() { + assertFalse(AsyncHttpClientConfigDefaults.defaultUseAbsoluteRequestDeadline()); + testBooleanSystemProperty("useAbsoluteRequestDeadline", "defaultUseAbsoluteRequestDeadline", "true"); + } + + @RepeatedIfExceptionsTest(repeats = 5) + public void testDefaultUseEventLoopTimeouts() { + assertFalse(AsyncHttpClientConfigDefaults.defaultUseEventLoopTimeouts()); + testBooleanSystemProperty("useEventLoopTimeouts", "defaultUseEventLoopTimeouts", "true"); + } + @RepeatedIfExceptionsTest(repeats = 5) public void testDefaultFollowRedirect() { assertFalse(AsyncHttpClientConfigDefaults.defaultFollowRedirect()); diff --git a/client/src/test/java/org/asynchttpclient/netty/timeout/TimeoutsHolderTest.java b/client/src/test/java/org/asynchttpclient/netty/timeout/TimeoutsHolderTest.java new file mode 100644 index 0000000000..ac67b5f1b9 --- /dev/null +++ b/client/src/test/java/org/asynchttpclient/netty/timeout/TimeoutsHolderTest.java @@ -0,0 +1,121 @@ +/* + * Copyright (c) 2026 AsyncHttpClient Project. All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.asynchttpclient.netty.timeout; + +import io.github.artsok.RepeatedIfExceptionsTest; +import org.asynchttpclient.AsyncCompletionHandler; +import org.asynchttpclient.AsyncHttpClientConfig; +import org.asynchttpclient.DefaultAsyncHttpClientConfig; +import org.asynchttpclient.Request; +import org.asynchttpclient.RequestBuilder; +import org.asynchttpclient.Response; +import org.asynchttpclient.channel.ChannelPoolPartitioning; +import org.asynchttpclient.netty.NettyResponseFuture; + +import java.time.Duration; + +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * The deadline a holder computes, which is where + * {@link AsyncHttpClientConfig#isUseAbsoluteRequestDeadline()} takes effect. A redirect, a retry and an auth + * replay each build a new holder for the same future, so what a second holder makes of the same exchange is + * the whole of the difference between the two modes. + *

+ * No timer and no request sender: without them the holder computes its deadline and arms nothing, which is + * exactly the part worth testing directly rather than through a request. + */ +public class TimeoutsHolderTest { + + private static final Duration BUDGET = Duration.ofMillis(600); + private static final long ELAPSED_MS = 100; + // The deadline is a wall-clock reading and the budget is netted off in whole milliseconds, so an anchored + // deadline lands within a few milliseconds of itself rather than exactly on it. + private static final long TOLERANCE_MS = 30; + + @RepeatedIfExceptionsTest(repeats = 5) + public void anAbsoluteDeadlineStaysWhereTheExchangeStarted() throws Exception { + NettyResponseFuture future = exchange(true); + + long firstHop = deadlineOf(future, BUDGET); + Thread.sleep(ELAPSED_MS); + long secondHop = deadlineOf(future, BUDGET); + + assertTrue(Math.abs(secondHop - firstHop) <= TOLERANCE_MS, + "the second hop moved the deadline by " + (secondHop - firstHop) + " ms"); + } + + @RepeatedIfExceptionsTest(repeats = 5) + public void aPerAttemptTimeoutGivesTheSecondHopItsOwnBudget() throws Exception { + NettyResponseFuture future = exchange(false); + + long firstHop = deadlineOf(future, BUDGET); + Thread.sleep(ELAPSED_MS); + long secondHop = deadlineOf(future, BUDGET); + + assertTrue(secondHop - firstHop >= ELAPSED_MS / 2, + "the second hop should have started a budget of its own, moved by only " + + (secondHop - firstHop) + " ms"); + } + + @RepeatedIfExceptionsTest(repeats = 5) + public void anExchangeThatOutranItsDeadlineHasNothingLeft() throws Exception { + // A budget this small is spent by the time the sleep is over, so the next hop has nothing to run in. + NettyResponseFuture future = exchange(true); + Thread.sleep(ELAPSED_MS); + + assertTrue(TimeoutsHolder.remainingBudget(config(Duration.ofMillis(1)), future) <= 0, + "a spent deadline should leave nothing to send a further hop with"); + } + + @RepeatedIfExceptionsTest(repeats = 5) + public void aPerAttemptExchangeIsNotBoundedAsAWhole() throws Exception { + // Asserted on the deadline the holder computes rather than on the budget: per attempt there is no + // exchange-wide budget to run out of, so the arithmetic is not what the answer rests on. + NettyResponseFuture future = exchange(false); + Thread.sleep(ELAPSED_MS); + + long deadline = deadlineOf(future, BUDGET); + + assertTrue(deadline - System.currentTimeMillis() >= BUDGET.toMillis() - TOLERANCE_MS, + "a hop should be given the configured timeout of its own however long the exchange has run, got " + + (deadline - System.currentTimeMillis()) + " ms"); + } + + private static long deadlineOf(NettyResponseFuture future, Duration requestTimeout) { + return holder(future, requestTimeout).requestTimeoutMillisTime(); + } + + private static TimeoutsHolder holder(NettyResponseFuture future, Duration requestTimeout) { + return new TimeoutsHolder(null, future, null, config(requestTimeout), null); + } + + private static AsyncHttpClientConfig config(Duration requestTimeout) { + return new DefaultAsyncHttpClientConfig.Builder().setRequestTimeout(requestTimeout).build(); + } + + private static NettyResponseFuture exchange(boolean useAbsoluteRequestDeadline) { + Request request = new RequestBuilder().setUrl("http://example.com:12345").build(); + NettyResponseFuture future = new NettyResponseFuture<>(request, new AsyncCompletionHandler() { + @Override + public Object onCompleted(Response response) { + return null; + } + }, null, 0, ChannelPoolPartitioning.PerHostChannelPoolPartitioning.INSTANCE, null, null); + future.setUseAbsoluteRequestDeadline(useAbsoluteRequestDeadline); + return future; + } +}