diff --git a/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java b/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java index 1104265d8..cae3900ee 100644 --- a/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java +++ b/client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java @@ -238,6 +238,27 @@ default LoadBalance getLoadBalance() { return LoadBalance.DEFAULT; } + /** + * Whether a recently-failed IP is briefly deprioritized when ordering a host's resolved addresses for a + * new connection. When enabled, a TCP connect failure to an address moves it to the back of the failover + * order for {@link #getFailedIpCooldownPeriod()} (it is never dropped, only re-ordered, and is re-probed + * once the window elapses). This applies regardless of {@link #getLoadBalance()} mode and bounds the cost + * of an IP that silently black-holes packets. + * + * @return {@code true} if the failed-IP cooldown is enabled + */ + default boolean isFailedIpCooldownEnabled() { + return true; + } + + /** + * @return how long a failed IP is deprioritized before it is re-probed; only used when + * {@link #isFailedIpCooldownEnabled()} is {@code true} + */ + default Duration getFailedIpCooldownPeriod() { + return Duration.ofSeconds(10); + } + /** * @return the disableUrlEncodingForBoundRequests */ diff --git a/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClientConfig.java b/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClientConfig.java index 5701e6ad7..38e4c9182 100644 --- a/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClientConfig.java +++ b/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClientConfig.java @@ -62,6 +62,8 @@ import static org.asynchttpclient.config.AsyncHttpClientConfigDefaults.defaultEnabledCipherSuites; 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.defaultFailedIpCooldownPeriod; import static org.asynchttpclient.config.AsyncHttpClientConfigDefaults.defaultFilterInsecureCipherSuites; import static org.asynchttpclient.config.AsyncHttpClientConfigDefaults.defaultFollowRedirect; import static org.asynchttpclient.config.AsyncHttpClientConfigDefaults.defaultHandshakeTimeout; @@ -133,6 +135,8 @@ public class DefaultAsyncHttpClientConfig implements AsyncHttpClientConfig { private final @Nullable Realm realm; private final int maxRequestRetry; private final LoadBalance loadBalance; + private final boolean failedIpCooldownEnabled; + private final Duration failedIpCooldownPeriod; private final boolean disableUrlEncodingForBoundRequests; private final boolean useLaxCookieEncoder; private final boolean disableZeroCopy; @@ -235,6 +239,8 @@ private DefaultAsyncHttpClientConfig(// http @Nullable Realm realm, int maxRequestRetry, LoadBalance loadBalance, + boolean failedIpCooldownEnabled, + Duration failedIpCooldownPeriod, boolean disableUrlEncodingForBoundRequests, boolean useLaxCookieEncoder, boolean disableZeroCopy, @@ -337,6 +343,8 @@ private DefaultAsyncHttpClientConfig(// http this.realm = realm; this.maxRequestRetry = maxRequestRetry; this.loadBalance = loadBalance; + this.failedIpCooldownEnabled = failedIpCooldownEnabled; + this.failedIpCooldownPeriod = failedIpCooldownPeriod; this.disableUrlEncodingForBoundRequests = disableUrlEncodingForBoundRequests; this.useLaxCookieEncoder = useLaxCookieEncoder; this.disableZeroCopy = disableZeroCopy; @@ -496,6 +504,16 @@ public LoadBalance getLoadBalance() { return loadBalance; } + @Override + public boolean isFailedIpCooldownEnabled() { + return failedIpCooldownEnabled; + } + + @Override + public Duration getFailedIpCooldownPeriod() { + return failedIpCooldownPeriod; + } + @Override public boolean isDisableUrlEncodingForBoundRequests() { return disableUrlEncodingForBoundRequests; @@ -908,6 +926,8 @@ public static class Builder { private @Nullable Realm realm; private int maxRequestRetry = defaultMaxRequestRetry(); private LoadBalance loadBalance = defaultLoadBalance(); + private boolean failedIpCooldownEnabled = defaultFailedIpCooldownEnabled(); + private Duration failedIpCooldownPeriod = defaultFailedIpCooldownPeriod(); private boolean disableUrlEncodingForBoundRequests = defaultDisableUrlEncodingForBoundRequests(); private boolean useLaxCookieEncoder = defaultUseLaxCookieEncoder(); private boolean disableZeroCopy = defaultDisableZeroCopy(); @@ -1013,6 +1033,8 @@ public Builder(AsyncHttpClientConfig config) { realm = config.getRealm(); maxRequestRetry = config.getMaxRequestRetry(); loadBalance = config.getLoadBalance(); + failedIpCooldownEnabled = config.isFailedIpCooldownEnabled(); + failedIpCooldownPeriod = config.getFailedIpCooldownPeriod(); disableUrlEncodingForBoundRequests = config.isDisableUrlEncodingForBoundRequests(); useLaxCookieEncoder = config.isUseLaxCookieEncoder(); disableZeroCopy = config.isDisableZeroCopy(); @@ -1184,6 +1206,31 @@ public Builder setLoadBalance(LoadBalance loadBalance) { return this; } + /** + * Enables or disables briefly deprioritizing a recently-failed IP when ordering a host's resolved + * addresses for a new connection (applies regardless of {@link #setLoadBalance(LoadBalance) load + * balancing} mode). + * + * @param failedIpCooldownEnabled whether the failed-IP cooldown is enabled + * @return this + * @see AsyncHttpClientConfig#isFailedIpCooldownEnabled() + */ + public Builder setFailedIpCooldownEnabled(boolean failedIpCooldownEnabled) { + this.failedIpCooldownEnabled = failedIpCooldownEnabled; + return this; + } + + /** + * @param failedIpCooldownPeriod how long a failed IP is deprioritized before it is re-probed; + * {@code null} resets to the default + * @return this + * @see AsyncHttpClientConfig#getFailedIpCooldownPeriod() + */ + public Builder setFailedIpCooldownPeriod(Duration failedIpCooldownPeriod) { + this.failedIpCooldownPeriod = failedIpCooldownPeriod == null ? defaultFailedIpCooldownPeriod() : failedIpCooldownPeriod; + return this; + } + public Builder setDisableUrlEncodingForBoundRequests(boolean disableUrlEncodingForBoundRequests) { this.disableUrlEncodingForBoundRequests = disableUrlEncodingForBoundRequests; return this; @@ -1660,6 +1707,8 @@ public DefaultAsyncHttpClientConfig build() { realm, maxRequestRetry, loadBalance, + failedIpCooldownEnabled, + failedIpCooldownPeriod, disableUrlEncodingForBoundRequests, useLaxCookieEncoder, disableZeroCopy, diff --git a/client/src/main/java/org/asynchttpclient/LoadBalance.java b/client/src/main/java/org/asynchttpclient/LoadBalance.java index 7caed9b67..c7f383642 100644 --- a/client/src/main/java/org/asynchttpclient/LoadBalance.java +++ b/client/src/main/java/org/asynchttpclient/LoadBalance.java @@ -70,16 +70,10 @@ public enum LoadBalance { * resolver that intentionally rotates its results, such as * {@link io.netty.resolver.RoundRobinInetAddressResolver} — that one is meant for * {@link #DEFAULT} mode, where it provides the spreading instead. - *
  • Rotation is liveness-aware only as a short-lived dampener, not a health checker. A failed - * connection attempt puts that IP in a brief cooldown, during which it is deprioritized (moved - * to the back of the failover order) before being re-probed once the window elapses. This bounds - * the cost of an IP that silently black-holes packets (drops them with no RST): such an IP would - * otherwise burn a full {@code connectTimeout} on every request pinned to it before TCP - * failover moved on to a healthy IP; with the cooldown only the occasional re-probe pays that - * cost. (An IP that actively refuses the connection fails over immediately and cheaply, with or - * without the cooldown.) The cooldown never removes an address from rotation — it only re-orders - * it — so authoritative liveness is still expected to be governed at the DNS/resolver level, as - * it already is in {@link #DEFAULT} mode.
    • + *
  • The rotation is applied on top of the failed-IP cooldown + * ({@link AsyncHttpClientConfig#isFailedIpCooldownEnabled()}), which briefly deprioritizes a + * recently-failed IP. That cooldown is a separate, mode-independent feature — it also applies in + * {@link #DEFAULT} mode — so it is documented on the config getter rather than here.
    • * */ ROUND_ROBIN diff --git a/client/src/main/java/org/asynchttpclient/config/AsyncHttpClientConfigDefaults.java b/client/src/main/java/org/asynchttpclient/config/AsyncHttpClientConfigDefaults.java index fd8de69b5..550381fbd 100644 --- a/client/src/main/java/org/asynchttpclient/config/AsyncHttpClientConfigDefaults.java +++ b/client/src/main/java/org/asynchttpclient/config/AsyncHttpClientConfigDefaults.java @@ -60,6 +60,8 @@ public final class AsyncHttpClientConfigDefaults { public static final String KEEP_ALIVE_CONFIG = "keepAlive"; 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 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"; public static final String USE_OPEN_SSL_CONFIG = "useOpenSsl"; @@ -171,6 +173,14 @@ public static boolean defaultEnableAutomaticDecompression() { return AsyncHttpClientConfigHelper.getAsyncHttpClientConfig().getBoolean(ASYNC_CLIENT_CONFIG_ROOT + ENABLE_AUTOMATIC_DECOMPRESSION_CONFIG); } + public static boolean defaultFailedIpCooldownEnabled() { + return AsyncHttpClientConfigHelper.getAsyncHttpClientConfig().getBoolean(ASYNC_CLIENT_CONFIG_ROOT + FAILED_IP_COOLDOWN_ENABLED_CONFIG); + } + + public static Duration defaultFailedIpCooldownPeriod() { + return AsyncHttpClientConfigHelper.getAsyncHttpClientConfig().getDuration(ASYNC_CLIENT_CONFIG_ROOT + FAILED_IP_COOLDOWN_PERIOD_CONFIG); + } + public static String defaultUserAgent() { return AsyncHttpClientConfigHelper.getAsyncHttpClientConfig().getString(ASYNC_CLIENT_CONFIG_ROOT + USER_AGENT_CONFIG); } diff --git a/client/src/main/java/org/asynchttpclient/netty/channel/FailedIpCooldownHolder.java b/client/src/main/java/org/asynchttpclient/netty/channel/FailedIpCooldownHolder.java new file mode 100644 index 000000000..58f79f96d --- /dev/null +++ b/client/src/main/java/org/asynchttpclient/netty/channel/FailedIpCooldownHolder.java @@ -0,0 +1,169 @@ +/* + * 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.channel; + +import java.net.InetSocketAddress; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ConcurrentHashMap; +import java.util.function.LongSupplier; + +/** + * Per-host failed-IP cooldown applied to a request's resolved addresses before a new connection is + * opened, independently of the configured {@link org.asynchttpclient.LoadBalance} mode. + * + *

    When a connection attempt to an address fails, {@link #markFailed(String, InetSocketAddress)} + * puts that address in a short cooldown. While the cooldown is active {@link #reorder(String, List)} + * moves the address to the back of the returned list rather than dropping it, so it is still + * available as a last-resort failover target and is re-probed once the window elapses. This bounds the + * cost of an IP that silently black-holes packets (drops them with no RST): without the cooldown every + * new connection targeting it would burn a full {@code connectTimeout} before failing over; with it, + * only the occasional re-probe pays that cost. (An IP that actively refuses the connection fails over + * immediately and cheaply, with or without the cooldown.) Liveness remains governed at the DNS/resolver + * level — the cooldown is only a short-lived dampener, not a health checker. + * + *

    The cooldown only re-orders the resolved addresses; it never removes one, so failover always has + * somewhere to go. It tracks TCP connect failures only (TLS/handshake failures are not fed back here), + * matching where address-level failover happens. + * + *

    Per-host state is held in a bounded map (capped at {@value #MAX_TRACKED_HOSTS}); at the cap an + * arbitrary entry is evicted before a new one is added, so memory stays bounded even for clients that + * touch very many distinct hosts. Dropping a host's state is harmless — it simply forgets any cooldowns + * the next time the host is seen. + * + *

    Thread-safe. + */ +public final class FailedIpCooldownHolder { + + // Cap on the number of per-host entries retained, so a client that touches very many distinct + // multi-IP hosts (crawler/gateway) can't grow this map without bound. At the cap an arbitrary + // entry is evicted before a new one is inserted (same approach as util/NonceCounter). + static final int MAX_TRACKED_HOSTS = 4096; + + // How long a failed address is deprioritized before it is re-probed. Deliberately coarser than the + // default connectTimeout (PT5S) so that a single failure actually routes traffic away from a dead IP + // for a useful window instead of re-pinning to it on the very next request, yet short enough that a + // recovered IP rejoins the order quickly. The DNS/resolver layer remains the authority on liveness. + static final Duration DEFAULT_FAILED_IP_COOLDOWN = Duration.ofSeconds(10); + + private final ConcurrentHashMap hosts = new ConcurrentHashMap<>(); + private final long cooldownNanos; + private final LongSupplier nanoClock; + + public FailedIpCooldownHolder() { + this(DEFAULT_FAILED_IP_COOLDOWN.toNanos(), System::nanoTime); + } + + public FailedIpCooldownHolder(long cooldownNanos, LongSupplier nanoClock) { + this.cooldownNanos = cooldownNanos; + this.nanoClock = nanoClock; + } + + /** + * Re-orders {@code addresses} so that any address currently in cooldown is moved to the back + * (otherwise preserving the incoming order). + * + * @param host the connection's target host (the key the matching {@link #markFailed} calls use) + * @param addresses the resolved socket addresses, in their incoming order + * @return the same list instance when there is nothing to do (size {@code <= 1}, or no address is in + * cooldown), otherwise a new list with the cooling addresses moved to the back + */ + public List reorder(String host, List addresses) { + if (addresses.size() <= 1) { + return addresses; + } + // Touch the per-host state even when nothing is cooling yet, so a subsequent markFailed for this + // host (which never resurrects an evicted entry) has somewhere to record the failure. + HostState state = stateFor(host); + if (state.cooldowns.isEmpty()) { + return addresses; + } + return moveCoolingToBack(state, addresses); + } + + /** + * Records that a connection attempt to {@code address} (for {@code host}) failed, so subsequent + * {@link #reorder} calls move it to the back for {@link #DEFAULT_FAILED_IP_COOLDOWN}. No-op when + * the host is not (or no longer) tracked — we never resurrect an evicted entry, which keeps the + * failure path from growing the map for hosts that are not actively being connected to. + */ + public void markFailed(String host, InetSocketAddress address) { + HostState state = hosts.get(host); + if (state != null) { + state.cooldowns.put(address, nanoClock.getAsLong() + cooldownNanos); + } + } + + // Visible for testing: the number of hosts currently tracked (bounded by MAX_TRACKED_HOSTS). + int trackedHostCount() { + return hosts.size(); + } + + private HostState stateFor(String host) { + evictIfNeeded(); + return hosts.computeIfAbsent(host, h -> new HostState()); + } + + // Stable-partition the order into not-cooling (kept first) and cooling (moved to the back), expiring + // elapsed cooldowns lazily as we go. If every address is cooling, the input is returned unchanged so we + // never hand back an empty list — failover still has somewhere to go. + private List moveCoolingToBack(HostState state, List addresses) { + long now = nanoClock.getAsLong(); + List healthy = new ArrayList<>(addresses.size()); + List cooling = null; + for (InetSocketAddress address : addresses) { + Long until = state.cooldowns.get(address); + if (until == null) { + healthy.add(address); + } else if (until - now > 0) { // nanoTime-safe comparison + if (cooling == null) { + cooling = new ArrayList<>(); + } + cooling.add(address); + } else { + state.cooldowns.remove(address, until); + healthy.add(address); + } + } + if (cooling == null) { + return addresses; // nothing actually cooling (all entries had expired) + } + if (healthy.isEmpty()) { + return addresses; // everything is cooling — keep the original order rather than return nothing + } + healthy.addAll(cooling); + return healthy; + } + + // Keep the map bounded: when it is full, drop one arbitrary entry before a new host is added. + // Evicting an entry only forgets that host's cooldowns, so the choice of victim does not matter. + private void evictIfNeeded() { + if (hosts.size() >= MAX_TRACKED_HOSTS) { + var it = hosts.keySet().iterator(); + if (it.hasNext()) { + it.next(); + it.remove(); + } + } + } + + // Per-host set of addresses currently in cooldown (address -> nanoTime the cooldown expires). The map + // is bounded by the host's resolved-IP count and self-prunes as entries expire during reorder. + private static final class HostState { + final ConcurrentHashMap cooldowns = new ConcurrentHashMap<>(); + } +} diff --git a/client/src/main/java/org/asynchttpclient/netty/channel/NettyChannelConnector.java b/client/src/main/java/org/asynchttpclient/netty/channel/NettyChannelConnector.java index 0c81c42f8..20a85b8ac 100644 --- a/client/src/main/java/org/asynchttpclient/netty/channel/NettyChannelConnector.java +++ b/client/src/main/java/org/asynchttpclient/netty/channel/NettyChannelConnector.java @@ -42,7 +42,7 @@ public class NettyChannelConnector { private final List remoteAddresses; private final AsyncHttpClientState clientState; // Notified with each remote address whose TCP connect attempt fails, or null when no caller cares. - // Used by round-robin load balancing to put a failed IP in a short cooldown; see RoundRobinAddressSelector. + // Used to put a failed IP in a short cooldown so new connections route around it; see FailedIpCooldownHolder. private final Consumer connectFailureListener; private volatile int i; @@ -104,7 +104,7 @@ public void onSuccess(Channel channel) { @Override public void onFailure(Channel channel, Throwable t) { if (connectFailureListener != null) { - // Record the failed IP before failing over so round-robin can route around it briefly. + // Record the failed IP before failing over so the cooldown can route around it briefly. connectFailureListener.accept(remoteAddress); } try { diff --git a/client/src/main/java/org/asynchttpclient/netty/channel/RoundRobinAddressSelector.java b/client/src/main/java/org/asynchttpclient/netty/channel/RoundRobinAddressSelector.java index f73c182d6..181294c21 100644 --- a/client/src/main/java/org/asynchttpclient/netty/channel/RoundRobinAddressSelector.java +++ b/client/src/main/java/org/asynchttpclient/netty/channel/RoundRobinAddressSelector.java @@ -16,12 +16,10 @@ package org.asynchttpclient.netty.channel; import java.net.InetSocketAddress; -import java.time.Duration; import java.util.ArrayList; import java.util.List; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicInteger; -import java.util.function.LongSupplier; /** * Picks, per host and per request, which resolved IP a new connection should target first when @@ -34,20 +32,14 @@ * requests only when the configured {@link io.netty.resolver.InetNameResolver} returns the * addresses in a stable order (see {@link org.asynchttpclient.LoadBalance#ROUND_ROBIN}). * - *

    Failed-IP cooldown. When a connection attempt to an address fails, {@link - * #markFailed(String, InetSocketAddress)} puts that address in a short cooldown. While the cooldown - * is active {@link #rotate(String, List)} deprioritizes the address — it is moved to the back of the - * returned list rather than dropped, so it is still available as a last-resort failover target and - * is re-probed once the window elapses. This bounds the cost of an IP that silently black-holes - * packets (no RST): without the cooldown every request pinned to it would burn a full - * {@code connectTimeout} before failing over; with it, only the occasional re-probe pays that cost. - * Liveness remains governed at the DNS/resolver level — the cooldown is only a short-lived dampener, - * not a health checker. + *

    This class is concerned only with rotation. Deprioritizing addresses whose connection attempts + * recently failed is handled separately and mode-independently by {@link FailedIpCooldownHolder}, applied + * on top of the rotation before a connection is opened. * *

    Per-host state is held in a bounded map (capped at {@value #MAX_TRACKED_HOSTS}); at the cap an * arbitrary entry is evicted before a new one is added, so memory stays bounded even for clients that * touch very many distinct hosts. Dropping a host's state is harmless — its rotation simply restarts - * at the first resolved address (and forgets any cooldowns) the next time it is seen. + * at the first resolved address the next time it is seen. * *

    Thread-safe. */ @@ -58,33 +50,14 @@ public final class RoundRobinAddressSelector { // entry is evicted before a new one is inserted (same approach as util/NonceCounter). static final int MAX_TRACKED_HOSTS = 4096; - // How long a failed address is deprioritized before it is re-probed. Deliberately coarser than the - // default connectTimeout (PT5S) so that a single failure actually routes traffic away from a dead IP - // for a useful window instead of re-pinning to it on the very next request, yet short enough that a - // recovered IP rejoins the rotation quickly. The DNS/resolver layer remains the authority on liveness. - static final Duration DEFAULT_FAILED_IP_COOLDOWN = Duration.ofSeconds(10); - - private final ConcurrentHashMap hosts = new ConcurrentHashMap<>(); - private final long cooldownNanos; - private final LongSupplier nanoClock; - - public RoundRobinAddressSelector() { - this(DEFAULT_FAILED_IP_COOLDOWN.toNanos(), System::nanoTime); - } - - // Visible for testing: lets tests drive a virtual clock and a custom cooldown deterministically. - RoundRobinAddressSelector(long cooldownNanos, LongSupplier nanoClock) { - this.cooldownNanos = cooldownNanos; - this.nanoClock = nanoClock; - } + private final ConcurrentHashMap counters = new ConcurrentHashMap<>(); /** * @param host the request's target host * @param resolved the resolved socket addresses (size {@code >= 1}), in resolver order * @return the same list instance when there is nothing to rotate (size {@code <= 1}, or the - * selected index is already first and no address is in cooldown), otherwise a new list whose first - * element is the round-robin-selected address, with any addresses currently in cooldown moved to the - * back (otherwise preserving resolver order) + * selected index is already first), otherwise a new list whose first element is the + * round-robin-selected address (otherwise preserving resolver order) */ public List rotate(String host, List resolved) { int n = resolved.size(); @@ -92,39 +65,19 @@ public List rotate(String host, List resol return resolved; } - HostState state = stateFor(host); - int index = (state.counter.getAndIncrement() & Integer.MAX_VALUE) % n; - - // Fast path: nothing failed recently, so the order is the plain round-robin rotation. - if (state.cooldowns.isEmpty()) { - return index == 0 ? resolved : rotateBy(resolved, index, n); - } - - List rotated = index == 0 ? resolved : rotateBy(resolved, index, n); - return deprioritizeCooling(state, rotated); - } - - /** - * Records that a connection attempt to {@code address} (for {@code host}) failed, so subsequent - * rotations deprioritize it for {@link #DEFAULT_FAILED_IP_COOLDOWN}. No-op when the host is not (or - * no longer) tracked — we never resurrect an evicted entry, which keeps the failure path from - * growing the map for hosts that round-robin is not actively rotating. - */ - public void markFailed(String host, InetSocketAddress address) { - HostState state = hosts.get(host); - if (state != null) { - state.cooldowns.put(address, nanoClock.getAsLong() + cooldownNanos); - } + AtomicInteger counter = counterFor(host); + int index = (counter.getAndIncrement() & Integer.MAX_VALUE) % n; + return index == 0 ? resolved : rotateBy(resolved, index, n); } // Visible for testing: the number of hosts currently tracked (bounded by MAX_TRACKED_HOSTS). int trackedHostCount() { - return hosts.size(); + return counters.size(); } - private HostState stateFor(String host) { + private AtomicInteger counterFor(String host) { evictIfNeeded(); - return hosts.computeIfAbsent(host, h -> new HostState()); + return counters.computeIfAbsent(host, h -> new AtomicInteger()); } private static List rotateBy(List resolved, int index, int n) { @@ -134,54 +87,15 @@ private static List rotateBy(List resolved return rotated; } - // Stable-partition the rotated order into not-cooling (kept first) and cooling (moved to the back), - // expiring elapsed cooldowns lazily as we go. If every address is cooling, the rotation is returned - // unchanged so we never hand back an empty list — failover still has somewhere to go. - private List deprioritizeCooling(HostState state, List rotated) { - long now = nanoClock.getAsLong(); - List healthy = new ArrayList<>(rotated.size()); - List cooling = null; - for (InetSocketAddress address : rotated) { - Long until = state.cooldowns.get(address); - if (until == null) { - healthy.add(address); - } else if (until - now > 0) { // nanoTime-safe comparison - if (cooling == null) { - cooling = new ArrayList<>(); - } - cooling.add(address); - } else { - state.cooldowns.remove(address, until); - healthy.add(address); - } - } - if (cooling == null) { - return rotated; // nothing actually cooling (all entries had expired) - } - if (healthy.isEmpty()) { - return rotated; // everything is cooling — keep the plain rotation rather than return nothing - } - healthy.addAll(cooling); - return healthy; - } - // Keep the map bounded: when it is full, drop one arbitrary entry before a new host is added. - // Evicting an entry only resets that host's rotation and cooldowns, so the choice of victim does not matter. + // Evicting an entry only resets that host's rotation, so the choice of victim does not matter. private void evictIfNeeded() { - if (hosts.size() >= MAX_TRACKED_HOSTS) { - var it = hosts.keySet().iterator(); + if (counters.size() >= MAX_TRACKED_HOSTS) { + var it = counters.keySet().iterator(); if (it.hasNext()) { it.next(); it.remove(); } } } - - // Per-host rotation cursor plus the set of addresses currently in cooldown (address -> nanoTime the - // cooldown expires). The cooldown map is bounded by the host's resolved-IP count and self-prunes as - // entries expire during rotation. - private static final class HostState { - final AtomicInteger counter = new AtomicInteger(); - final ConcurrentHashMap cooldowns = new ConcurrentHashMap<>(); - } } 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 a92789fb7..37b5061f8 100755 --- a/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java +++ b/client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java @@ -66,6 +66,7 @@ import org.asynchttpclient.netty.channel.ConnectionSemaphore; import org.asynchttpclient.netty.channel.Http2ConnectionState; import org.asynchttpclient.netty.channel.DefaultConnectionSemaphoreFactory; +import org.asynchttpclient.netty.channel.FailedIpCooldownHolder; import org.asynchttpclient.netty.channel.NettyChannelConnector; import org.asynchttpclient.netty.channel.NettyConnectListener; import org.asynchttpclient.netty.channel.RoundRobinAddressSelector; @@ -115,6 +116,9 @@ public final class NettyRequestSender { private final AsyncHttpClientState clientState; private final NettyRequestFactory requestFactory; private final RoundRobinAddressSelector rrSelector = new RoundRobinAddressSelector(); + // Deprioritizes a recently-failed IP when ordering a direct connection's resolved addresses, in any + // LoadBalance mode. Null when the failed-IP cooldown is disabled; call sites gate on ipCooldown != null. + private final FailedIpCooldownHolder ipCooldown; public NettyRequestSender(AsyncHttpClientConfig config, ChannelManager channelManager, Timer nettyTimer, AsyncHttpClientState clientState) { this.config = config; @@ -125,6 +129,9 @@ public NettyRequestSender(AsyncHttpClientConfig config, ChannelManager channelMa this.nettyTimer = nettyTimer; this.clientState = clientState; requestFactory = new NettyRequestFactory(config); + ipCooldown = config.isFailedIpCooldownEnabled() + ? new FailedIpCooldownHolder(config.getFailedIpCooldownPeriod().toNanos(), System::nanoTime) + : null; } // needConnect returns true if the request is secure/websocket and a HTTP proxy is set @@ -149,7 +156,7 @@ public ListenableFuture sendRequest(final Request request, final AsyncHan if (config.getLoadBalance() == LoadBalance.ROUND_ROBIN) { boolean overrideMatchesBase = future != null && future.getRoundRobinBaseUri() != null && request.getUri().isSameBase(future.getRoundRobinBaseUri()); - if (isRoundRobinEligible(request, proxyServer) && !overrideMatchesBase) { + if (isDirectConnection(request, proxyServer) && !overrideMatchesBase) { return sendRequestRoundRobin(request, asyncHandler, future, proxyServer); } if (!overrideMatchesBase && future != null && future.getRoundRobinBaseUri() != null) { @@ -176,12 +183,13 @@ public ListenableFuture sendRequest(final Request request, final AsyncHan } } - // A request is eligible for round-robin only when it opens a direct connection to the target host - // (the connector targets the resolved IPs). Excluded: explicit address (bypasses resolution), and - // any proxied host — HTTP or SOCKS — since the socket is established to the proxy rather than - // directly to the rotated target IPs. Round-robin still applies when the proxy is bypassed for - // the host (isIgnoredForHost), because that request connects directly. - private boolean isRoundRobinEligible(Request request, ProxyServer proxyServer) { + // Whether the request opens a direct connection to the target host, i.e. the connector targets the + // host's resolved IPs (keyed in DNS/cooldown state by uri.getHost()). Excluded: an explicit address + // (bypasses resolution), and any proxied host — HTTP or SOCKS — since the socket is established to the + // proxy rather than to the resolved target IPs. A bypassed proxy (isIgnoredForHost) still connects + // directly. Gates both round-robin rotation and the failed-IP cooldown so both stay keyed on the + // host whose IPs are actually being connected to. + private boolean isDirectConnection(Request request, ProxyServer proxyServer) { if (request.getAddress() != null || needConnect(request, proxyServer)) { return false; } @@ -213,6 +221,11 @@ protected void onSuccess(List addresses) { List ordered = addresses; if (addresses.size() > 1) { ordered = rrSelector.rotate(host, addresses); + // Apply the failed-IP cooldown on top of the rotation, before pinning the IP-aware + // partition key below, so the pool pin and the chosen IP avoid a recently-dead address. + if (ipCooldown != null) { + ordered = ipCooldown.reorder(host, ordered); + } InetAddress chosen = ordered.get(0).getAddress(); Object baseKey = request.getChannelPoolPartitioning().getPartitionKey(uri, request.getVirtualHost(), proxyServer); newFuture.setPartitionKeyOverride(new RoundRobinPartitionKey(baseKey, chosen)); @@ -458,7 +471,15 @@ private ListenableFuture sendRequestWithNewChannel(Request request, Proxy @Override protected void onSuccess(List addresses) { - connectWithAddresses(request, proxy, future, asyncHandler, addresses); + List ordered = addresses; + // Apply the failed-IP cooldown to direct connections regardless of LoadBalance mode, so a + // recently-failed IP is deprioritized on the next new connection. Skipped for the + // round-robin reuse branch above (those addresses are already cooldown-ordered) and for + // proxied/explicit-address requests (the resolved addresses aren't the target host's IPs). + if (ipCooldown != null && addresses.size() > 1 && isDirectConnection(request, proxy)) { + ordered = ipCooldown.reorder(request.getUri().getHost(), addresses); + } + connectWithAddresses(request, proxy, future, asyncHandler, ordered); } @Override @@ -473,12 +494,13 @@ protected void onFailure(Throwable cause) { private void connectWithAddresses(Request request, ProxyServer proxy, NettyResponseFuture future, AsyncHandler asyncHandler, List addresses) { NettyConnectListener connectListener = new NettyConnectListener<>(future, NettyRequestSender.this, channelManager, connectionSemaphore); - // In round-robin mode, feed TCP connect failures back so the selector deprioritizes a dead IP for a - // short cooldown instead of re-pinning the next request to it (and burning another connectTimeout). + // Feed TCP connect failures back so the cooldown deprioritizes a dead IP for a short window instead + // of the next new connection re-targeting it (and burning another connectTimeout). Applied to direct + // connections in any LoadBalance mode; the host key matches the one reorder() ordered under. Consumer connectFailureListener = null; - if (future.getPartitionKeyOverride() instanceof RoundRobinPartitionKey) { + if (ipCooldown != null && isDirectConnection(request, proxy)) { String host = request.getUri().getHost(); - connectFailureListener = address -> rrSelector.markFailed(host, address); + connectFailureListener = address -> ipCooldown.markFailed(host, address); } NettyChannelConnector connector = new NettyChannelConnector(request.getLocalAddress(), addresses, asyncHandler, clientState, connectFailureListener); if (!future.isDone()) { 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 23d7b9985..5df97add5 100644 --- a/client/src/main/resources/org/asynchttpclient/config/ahc-default.properties +++ b/client/src/main/resources/org/asynchttpclient/config/ahc-default.properties @@ -24,6 +24,8 @@ org.asynchttpclient.strict302Handling=false org.asynchttpclient.keepAlive=true org.asynchttpclient.maxRequestRetry=5 org.asynchttpclient.loadBalance=DEFAULT +org.asynchttpclient.failedIpCooldownEnabled=true +org.asynchttpclient.failedIpCooldownPeriod=PT10S org.asynchttpclient.disableUrlEncodingForBoundRequests=false org.asynchttpclient.useLaxCookieEncoder=false org.asynchttpclient.removeQueryParamOnRedirect=true diff --git a/client/src/test/java/org/asynchttpclient/FailedIpCooldownConfigTest.java b/client/src/test/java/org/asynchttpclient/FailedIpCooldownConfigTest.java new file mode 100644 index 000000000..94e7b7c34 --- /dev/null +++ b/client/src/test/java/org/asynchttpclient/FailedIpCooldownConfigTest.java @@ -0,0 +1,63 @@ +/* + * 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 org.junit.jupiter.api.Test; + +import java.time.Duration; + +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.assertTrue; + +class FailedIpCooldownConfigTest { + + @Test + void defaultsToEnabledWithTenSecondPeriod() { + AsyncHttpClientConfig config = config().build(); + assertTrue(config.isFailedIpCooldownEnabled()); + assertEquals(Duration.ofSeconds(10), config.getFailedIpCooldownPeriod()); + } + + @Test + void builderSetsEnabled() { + assertFalse(config().setFailedIpCooldownEnabled(false).build().isFailedIpCooldownEnabled()); + } + + @Test + void builderSetsPeriod() { + AsyncHttpClientConfig config = config().setFailedIpCooldownPeriod(Duration.ofSeconds(30)).build(); + assertEquals(Duration.ofSeconds(30), config.getFailedIpCooldownPeriod()); + } + + @Test + void nullPeriodResetsToDefault() { + AsyncHttpClientConfig config = config().setFailedIpCooldownPeriod(null).build(); + assertEquals(Duration.ofSeconds(10), config.getFailedIpCooldownPeriod()); + } + + @Test + void copyConstructorPreservesValues() { + AsyncHttpClientConfig source = config() + .setFailedIpCooldownEnabled(false) + .setFailedIpCooldownPeriod(Duration.ofSeconds(42)) + .build(); + AsyncHttpClientConfig copy = new DefaultAsyncHttpClientConfig.Builder(source).build(); + assertFalse(copy.isFailedIpCooldownEnabled()); + assertEquals(Duration.ofSeconds(42), copy.getFailedIpCooldownPeriod()); + } +} diff --git a/client/src/test/java/org/asynchttpclient/netty/channel/FailedIpCooldownHolderTest.java b/client/src/test/java/org/asynchttpclient/netty/channel/FailedIpCooldownHolderTest.java new file mode 100644 index 000000000..454676fe0 --- /dev/null +++ b/client/src/test/java/org/asynchttpclient/netty/channel/FailedIpCooldownHolderTest.java @@ -0,0 +1,129 @@ +/* + * 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.channel; + +import org.junit.jupiter.api.Test; + +import java.net.InetSocketAddress; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.function.LongSupplier; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class FailedIpCooldownHolderTest { + + private static InetSocketAddress addr(String ip) { + return new InetSocketAddress(ip, 80); + } + + private static String firstIp(List addresses) { + return addresses.get(0).getAddress().getHostAddress(); + } + + @Test + void failedAddressIsDeprioritizedDuringCooldown() { + long[] now = {1_000}; + LongSupplier clock = () -> now[0]; + // cooldown of 100 ticks, driven by a virtual clock so the test is deterministic + FailedIpCooldownHolder cooldown = new FailedIpCooldownHolder(100, clock); + List input = Arrays.asList(addr("127.0.0.1"), addr("127.0.0.2")); + + cooldown.reorder("h", input); // create per-host state so markFailed is recorded + cooldown.markFailed("h", addr("127.0.0.1")); // 127.0.0.1 enters cooldown until tick 1100 + + // the cooling 127.0.0.1 is moved to the back; the healthy 127.0.0.2 comes first + List ordered = cooldown.reorder("h", input); + assertEquals("127.0.0.2", firstIp(ordered)); + assertTrue(ordered.containsAll(input), "the cooling address is kept as a last-resort failover target"); + } + + @Test + void cooledAddressIsReprobedAfterWindowElapses() { + long[] now = {1_000}; + LongSupplier clock = () -> now[0]; + FailedIpCooldownHolder cooldown = new FailedIpCooldownHolder(100, clock); + List input = Arrays.asList(addr("127.0.0.1"), addr("127.0.0.2")); + + cooldown.reorder("h", input); // create state + cooldown.markFailed("h", addr("127.0.0.1")); // cooldown until tick 1100 + now[0] = 1_201; // advance well past the cooldown window + + // once the window has elapsed the address rejoins the order in its original position + assertEquals("127.0.0.1", firstIp(cooldown.reorder("h", input)), + "a recovered IP must be re-probed after its cooldown elapses"); + } + + @Test + void allAddressesCoolingFallsBackToOriginalOrder() { + long[] now = {1_000}; + LongSupplier clock = () -> now[0]; + FailedIpCooldownHolder cooldown = new FailedIpCooldownHolder(100, clock); + List input = Arrays.asList(addr("127.0.0.1"), addr("127.0.0.2")); + + cooldown.reorder("h", input); + cooldown.markFailed("h", addr("127.0.0.1")); + cooldown.markFailed("h", addr("127.0.0.2")); + + // every address is cooling — must still hand back the full list, never an empty one + List ordered = cooldown.reorder("h", input); + assertEquals(2, ordered.size()); + assertTrue(ordered.containsAll(input)); + } + + @Test + void markFailedForUntrackedHostIsNoOp() { + FailedIpCooldownHolder cooldown = new FailedIpCooldownHolder(100, () -> 0L); + // a host that was never reordered must not be resurrected (or have memory allocated) by a failure + cooldown.markFailed("never-reordered", addr("127.0.0.1")); + assertEquals(0, cooldown.trackedHostCount()); + } + + @Test + void singleAddressReturnedUnchangedAndUntracked() { + FailedIpCooldownHolder cooldown = new FailedIpCooldownHolder(100, () -> 0L); + List input = Collections.singletonList(addr("127.0.0.1")); + // size <= 1 short-circuits: same instance back, and no per-host state allocated + assertSame(input, cooldown.reorder("h", input)); + assertEquals(0, cooldown.trackedHostCount()); + } + + @Test + void reorderWithoutFailuresReturnsInputUnchanged() { + FailedIpCooldownHolder cooldown = new FailedIpCooldownHolder(100, () -> 0L); + List input = Arrays.asList(addr("127.0.0.1"), addr("127.0.0.2")); + // nothing has failed yet, so the order is preserved (same instance returned on the fast path) + assertSame(input, cooldown.reorder("h", input)); + } + + @Test + void boundsTrackedHosts() { + FailedIpCooldownHolder cooldown = new FailedIpCooldownHolder(100, () -> 0L); + List input = Arrays.asList(addr("127.0.0.1"), addr("127.0.0.2")); + + // Drive far more distinct hosts through the cooldown than the cap allows; an arbitrary entry is + // evicted before each new host is added once the map is full, so the tracker stays bounded. + for (int i = 0; i < FailedIpCooldownHolder.MAX_TRACKED_HOSTS * 3; i++) { + cooldown.reorder("host-" + i, input); + } + + assertTrue(cooldown.trackedHostCount() <= FailedIpCooldownHolder.MAX_TRACKED_HOSTS, + "tracked hosts must stay bounded by the cap"); + } +} diff --git a/client/src/test/java/org/asynchttpclient/netty/channel/RoundRobinAddressSelectorTest.java b/client/src/test/java/org/asynchttpclient/netty/channel/RoundRobinAddressSelectorTest.java index 09f3ff3c7..8bb09033f 100644 --- a/client/src/test/java/org/asynchttpclient/netty/channel/RoundRobinAddressSelectorTest.java +++ b/client/src/test/java/org/asynchttpclient/netty/channel/RoundRobinAddressSelectorTest.java @@ -23,7 +23,6 @@ import java.util.HashMap; import java.util.List; import java.util.Map; -import java.util.function.LongSupplier; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotEquals; @@ -108,67 +107,4 @@ void perHostCountersAreIndependent() { assertEquals("127.0.0.2", firstIp(selector.rotate("a", input))); assertNotEquals(firstIp(selector.rotate("b", input)), "127.0.0.1"); } - - @Test - void failedAddressIsDeprioritizedDuringCooldown() { - long[] now = {1_000}; - LongSupplier clock = () -> now[0]; - // cooldown of 100 ticks, driven by a virtual clock so the test is deterministic - RoundRobinAddressSelector selector = new RoundRobinAddressSelector(100, clock); - List input = Arrays.asList(addr("127.0.0.1"), addr("127.0.0.2")); - - selector.rotate("h", input); // index 0 -> 127.0.0.1 first, counter now 1 - selector.markFailed("h", addr("127.0.0.1")); // 127.0.0.1 enters cooldown until tick 1100 - - // index 1 -> 127.0.0.2 naturally first - assertEquals("127.0.0.2", firstIp(selector.rotate("h", input))); - // index 0 would put the cooling 127.0.0.1 first, but it is deprioritized to the back - List rotated = selector.rotate("h", input); - assertEquals("127.0.0.2", firstIp(rotated)); - assertTrue(rotated.containsAll(input), "the cooling address is kept as a last-resort failover target"); - } - - @Test - void cooledAddressIsReprobedAfterWindowElapses() { - long[] now = {1_000}; - LongSupplier clock = () -> now[0]; - RoundRobinAddressSelector selector = new RoundRobinAddressSelector(100, clock); - List input = Arrays.asList(addr("127.0.0.1"), addr("127.0.0.2")); - - selector.rotate("h", input); // counter now 1 - selector.markFailed("h", addr("127.0.0.1")); // cooldown until tick 1100 - now[0] = 1_201; // advance well past the cooldown window - - // once the window has elapsed the address rejoins the plain rotation and can be selected first again - boolean reprobed = false; - for (int i = 0; i < 2 && !reprobed; i++) { - reprobed = "127.0.0.1".equals(firstIp(selector.rotate("h", input))); - } - assertTrue(reprobed, "a recovered IP must be re-probed after its cooldown elapses"); - } - - @Test - void allAddressesCoolingFallsBackToPlainRotation() { - long[] now = {1_000}; - LongSupplier clock = () -> now[0]; - RoundRobinAddressSelector selector = new RoundRobinAddressSelector(100, clock); - List input = Arrays.asList(addr("127.0.0.1"), addr("127.0.0.2")); - - selector.rotate("h", input); - selector.markFailed("h", addr("127.0.0.1")); - selector.markFailed("h", addr("127.0.0.2")); - - // every address is cooling — rotation must still hand back the full list, never an empty one - List rotated = selector.rotate("h", input); - assertEquals(2, rotated.size()); - assertTrue(rotated.containsAll(input)); - } - - @Test - void markFailedForUntrackedHostIsNoOp() { - RoundRobinAddressSelector selector = new RoundRobinAddressSelector(); - // a host that was never rotated must not be resurrected (or have memory allocated) by a failure - selector.markFailed("never-rotated", addr("127.0.0.1")); - assertEquals(0, selector.trackedHostCount()); - } }