Fix the defects the rerunner removal uncovered

.github/workflows/builds.yml

@@ -7,6 +7,7 @@ on:
 jobs:
   Verify:
     runs-on: ubuntu-latest
+    timeout-minutes: 15
     steps:
       - uses: actions/checkout@v7
       - uses: actions/setup-java@v6.0.1
@@ -20,6 +21,7 @@ jobs:
 
   RunOnLinux:
     runs-on: ubuntu-latest
+    timeout-minutes: 45
     needs: Verify
     steps:
       - uses: actions/checkout@v7
@@ -34,6 +36,7 @@ jobs:
 
   RunOnMacOs:
     runs-on: macos-latest
+    timeout-minutes: 45
     needs: Verify
     steps:
       - uses: actions/checkout@v7
@@ -48,6 +51,7 @@ jobs:
 
   RunOnWindows:
     runs-on: windows-latest
+    timeout-minutes: 45
     needs: Verify
     steps:
       - uses: actions/checkout@v7

.github/workflows/maven.yml

@@ -21,6 +21,7 @@ on:
 jobs:
   compile-and-check:
     runs-on: ubuntu-latest
+    timeout-minutes: 15
     steps:
       - uses: actions/checkout@v7
       # The last entry wins the default JAVA_HOME, so the baseline check runs on JDK 11.
@@ -62,6 +63,8 @@ jobs:
             distribution: corretto
 
     runs-on: ${{ matrix.os }}
+    # A test job takes under 10 minutes. Nothing retries a hung test any more, so bound it here.
+    timeout-minutes: 30
     steps:
       - uses: actions/checkout@v7
       - uses: actions/setup-java@v6.0.1

client/src/main/java/org/asynchttpclient/netty/channel/Channels.java

@@ -28,6 +28,7 @@ public final class Channels {
 
     private static final AttributeKey<Object> DEFAULT_ATTRIBUTE = AttributeKey.valueOf("default");
     private static final AttributeKey<Active> ACTIVE_TOKEN_ATTRIBUTE = AttributeKey.valueOf("activeToken");
+    private static final AttributeKey<Runnable> PERMIT_RELEASE_ATTRIBUTE = AttributeKey.valueOf("permitRelease");
 
     private Channels() {
         // Prevent outside initialization
@@ -58,6 +59,21 @@ public static boolean isActiveTokenSet(Channel channel) {
         return channel != null && channel.attr(ACTIVE_TOKEN_ATTRIBUTE).getAndSet(null) != null;
     }
 
+    static void setPermitRelease(Channel channel, Runnable release) {
+        channel.attr(PERMIT_RELEASE_ATTRIBUTE).set(release);
+    }
+
+    /**
+     * Releases the connection permit this channel holds, if any. Safe to call more than once; the close
+     * listener remains the backstop.
+     */
+    public static void releasePermit(Channel channel) {
+        Runnable release = channel.attr(PERMIT_RELEASE_ATTRIBUTE).get();
+        if (release != null) {
+            release.run();
+        }
+    }
+
     public static void silentlyCloseChannel(Channel channel) {
         try {
             if (channel != null && channel.isActive()) {

client/src/main/java/org/asynchttpclient/netty/channel/CombinedConnectionSemaphore.java

@@ -18,16 +18,26 @@
 import java.io.IOException;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
+import java.util.function.LongSupplier;
 
 /**
  * A combined {@link ConnectionSemaphore} with two limits - a global limit and a per-host limit
  */
 public class CombinedConnectionSemaphore extends PerHostConnectionSemaphore {
     protected final MaxConnectionSemaphore globalMaxConnectionSemaphore;
 
+    private final LongSupplier millisClock;
+
     CombinedConnectionSemaphore(int maxConnections, int maxConnectionsPerHost, int acquireTimeout) {
+        // Monotonic, and finer than the 15 ms steps currentTimeMillis takes on Windows.
+        this(maxConnections, maxConnectionsPerHost, acquireTimeout, () -> TimeUnit.NANOSECONDS.toMillis(System.nanoTime()));
+    }
+
+    // For tests: the split of the budget can be checked without waiting for it.
+    CombinedConnectionSemaphore(int maxConnections, int maxConnectionsPerHost, int acquireTimeout, LongSupplier millisClock) {
         super(maxConnectionsPerHost, acquireTimeout);
         globalMaxConnectionSemaphore = new MaxConnectionSemaphore(maxConnections, acquireTimeout);
+        this.millisClock = millisClock;
     }
 
     @Override
@@ -89,9 +99,9 @@ protected long acquireGlobal(Object partitionKey) throws IOException {
      * Acquires the global lock and returns the remaining time, in millis, to acquire the per-host lock
      */
     protected long acquireGlobalTimed(Object partitionKey) throws IOException {
-        long beforeGlobalAcquire = System.currentTimeMillis();
+        long beforeGlobalAcquire = millisClock.getAsLong();
         acquireGlobal(partitionKey);
-        long lockTime = System.currentTimeMillis() - beforeGlobalAcquire;
+        long lockTime = millisClock.getAsLong() - beforeGlobalAcquire;
         return acquireTimeout - lockTime;
     }
 

client/src/main/java/org/asynchttpclient/netty/channel/NettyConnectListener.java

@@ -47,6 +47,8 @@ public final class NettyConnectListener<T> {
     private final NettyResponseFuture<T> future;
     private final ChannelManager channelManager;
     private final ConnectionSemaphore connectionSemaphore;
+    // Shared by the addresses one connect fails over, but only the one that connects reaches writeRequest.
+    private volatile boolean requestWritten;
 
     public NettyConnectListener(NettyResponseFuture<T> future, NettyRequestSender requestSender, ChannelManager channelManager, ConnectionSemaphore connectionSemaphore) {
         this.future = future;
@@ -83,6 +85,7 @@ private void writeRequest(Channel channel) {
         Channels.setAttribute(channel, future);
 
         channelManager.registerOpenChannel(channel);
+        requestWritten = true;
         requestSender.writeRequest(future, channel);
     }
 
@@ -97,6 +100,8 @@ public void onSuccess(Channel channel, InetSocketAddress remoteAddress) {
         final Object partitionKeyLock = semaphore != null ? future.takePartitionKeyLock() : null;
         final AtomicReference<Object> permit = new AtomicReference<>(partitionKeyLock);
         if (partitionKeyLock != null) {
+            // Also reachable from the channel, so an abort can return the permit before it tells the handler.
+            Channels.setPermitRelease(channel, () -> releasePermitOnce(semaphore, permit));
             channel.closeFuture().addListener(f -> releasePermitOnce(semaphore, permit));
         }
 
@@ -351,8 +356,7 @@ private void registerHttp2AndManageSemaphore(Channel channel, ConnectionSemaphor
     }
 
     /**
-     * Must only be called before {@link #writeRequest}: it may replay the request, and replaying one that was
-     * already written would send it twice.
+     * Replays the request only while it is unwritten: replaying one that was already written would send it twice.
      */
     public void onFailure(Channel channel, Throwable cause) {
 
@@ -362,6 +366,7 @@ public void onFailure(Channel channel, Throwable cause) {
         boolean canRetry = future.incrementRetryAndCheck();
         LOGGER.debug("Trying to recover from failing to connect channel {} with a retry value of {} ", channel, canRetry);
         if (canRetry//
+                && !requestWritten
                 && cause != null // FIXME when can we have a null cause?
                 && (future.getChannelState() != ChannelState.NEW || StackTraceInspector.recoverOnNettyDisconnectException(cause))) {
 

client/src/main/java/org/asynchttpclient/netty/future/StackTraceInspector.java

@@ -19,6 +19,7 @@
 
 import java.io.IOException;
 import java.net.ConnectException;
+import java.net.NoRouteToHostException;
 import java.nio.channels.ClosedChannelException;
 
 public final class StackTraceInspector {
@@ -45,9 +46,11 @@ private static boolean recoverOnConnectCloseException(Throwable t) {
             if (t instanceof ConnectTimeoutException) {
                 return false;
             }
-            // The type covers every transport. The frames (checkConnect up to JDK 12, pollConnect after)
-            // still matter: NIO reports an unreachable peer as NoRouteToHostException, not a ConnectException.
+            // The types cover every transport: native ones report an unreachable peer as a bare
+            // NoRouteToHostException. The frames (checkConnect up to JDK 12, pollConnect after) keep whatever
+            // else NIO throws from connect completion.
             if (t instanceof ConnectException
+                    || t instanceof NoRouteToHostException
                     || exceptionInMethod(t, "sun.nio.ch.SocketChannelImpl", "checkConnect")
                     || exceptionInMethod(t, "sun.nio.ch.Net", "pollConnect")) {
                 return true;

client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java

@@ -1155,6 +1155,13 @@ public void abort(Channel channel, NettyResponseFuture<?> future, Throwable t) {
         // NettyConnectListener.onFailure to abort the same future with a ConnectException instead -- which a
         // request timeout on the connect path can now hit, since the channel is published before the
         // handshake (issue #2189). The close still uses the channel passed in, which abort() does not clear.
+        //
+        // The permit goes back first: the channel is closed below either way, and a handler that sends its
+        // next request from onThrowable must not be refused by the connection it is being told has failed.
+        // An HTTP/2 connection keeps its permit until it stops serving streams.
+        if (channel != null && !ChannelManager.isHttp2(channel)) {
+            Channels.releasePermit(channel);
+        }
         if (!future.isDone()) {
             future.setChannelState(ChannelState.CLOSED);
             LOGGER.debug("Aborting Future {}\n", future);
@@ -1192,8 +1199,8 @@ public boolean retry(NettyResponseFuture<?> future) {
             try {
                 future.getAsyncHandler().onRetry();
             } catch (Exception e) {
+                // Throwing is how a handler refuses a retry. The caller aborts with the failure it has.
                 LOGGER.error("onRetry crashed", e);
-                abort(future.channel(), future, e);
                 return false;
             }
 

client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java

@@ -22,6 +22,7 @@
 import org.asynchttpclient.Request;
 import org.asynchttpclient.netty.NettyResponseFuture;
 import org.asynchttpclient.netty.request.NettyRequestSender;
+import org.asynchttpclient.util.DateUtils;
 import org.jetbrains.annotations.Nullable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -30,8 +31,7 @@
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
-
-import static org.asynchttpclient.util.DateUtils.unpreciseMillisTime;
+import java.util.function.LongSupplier;
 
 /**
  * The request and read timeouts of one exchange, armed either on the client's {@link Timer} or on the event
@@ -56,6 +56,8 @@ public class TimeoutsHolder {
     private volatile @Nullable ReadTimeoutTimerTask readTimeoutTask;
     private final NettyResponseFuture<?> nettyResponseFuture;
     private volatile InetSocketAddress remoteAddress;
+    private final LongSupplier millisClock;
+    private final LongSupplier nanoClock;
 
     public TimeoutsHolder(Timer nettyTimer, NettyResponseFuture<?> nettyResponseFuture, NettyRequestSender requestSender,
                           AsyncHttpClientConfig config, InetSocketAddress originalRemoteAddress) {
@@ -71,6 +73,17 @@ public TimeoutsHolder(Timer nettyTimer, NettyResponseFuture<?> nettyResponseFutu
      */
     public TimeoutsHolder(Timer nettyTimer, @Nullable EventExecutor eventExecutor, NettyResponseFuture<?> nettyResponseFuture,
                           NettyRequestSender requestSender, AsyncHttpClientConfig config, InetSocketAddress originalRemoteAddress) {
+        this(nettyTimer, eventExecutor, nettyResponseFuture, requestSender, config, originalRemoteAddress,
+                DateUtils::unpreciseMillisTime, System::nanoTime);
+    }
+
+    // For tests: a deadline can be checked without letting time pass. nanoClock must read the clock the
+    // future took its start from.
+    TimeoutsHolder(Timer nettyTimer, @Nullable EventExecutor eventExecutor, NettyResponseFuture<?> nettyResponseFuture,
+                   NettyRequestSender requestSender, AsyncHttpClientConfig config, InetSocketAddress originalRemoteAddress,
+                   LongSupplier millisClock, LongSupplier nanoClock) {
+        this.millisClock = millisClock;
+        this.nanoClock = nanoClock;
         this.nettyTimer = nettyTimer;
         this.eventExecutor = eventExecutor;
         this.nettyResponseFuture = nettyResponseFuture;
@@ -93,8 +106,8 @@ public TimeoutsHolder(Timer nettyTimer, @Nullable EventExecutor eventExecutor, N
             // 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);
+            requestTimeoutMillisTime = millisClock.getAsLong()
+                    + (absoluteDeadline ? remainingBudget(requestTimeoutInMs, nettyResponseFuture, nanoClock) : requestTimeoutInMs);
             requestTimeoutTask = new RequestTimeoutTimerTask(nettyResponseFuture, requestSender, this, requestTimeoutInMs);
         } else {
             requestTimeoutMillisTime = -1L;
@@ -125,7 +138,7 @@ public void start() {
             // 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);
+                    ? Math.max(remainingBudget(requestTimeoutValue, nettyResponseFuture, nanoClock), 0L) : requestTimeoutValue);
         }
     }
 
@@ -142,17 +155,21 @@ public void start() {
      * @see org.asynchttpclient.AsyncHttpClientConfig#isUseAbsoluteRequestDeadline()
      */
     public static long remainingBudget(AsyncHttpClientConfig config, NettyResponseFuture<?> nettyResponseFuture) {
+        return remainingBudget(config, nettyResponseFuture, System::nanoTime);
+    }
+
+    static long remainingBudget(AsyncHttpClientConfig config, NettyResponseFuture<?> nettyResponseFuture, LongSupplier nanoClock) {
         if (!nettyResponseFuture.isUseAbsoluteRequestDeadline()) {
             return Long.MAX_VALUE;
         }
-        return remainingBudget(requestTimeout(config, nettyResponseFuture.getTargetRequest()), nettyResponseFuture);
+        return remainingBudget(requestTimeout(config, nettyResponseFuture.getTargetRequest()), nettyResponseFuture, nanoClock);
     }
 
-    private static long remainingBudget(long requestTimeoutInMs, NettyResponseFuture<?> nettyResponseFuture) {
+    private static long remainingBudget(long requestTimeoutInMs, NettyResponseFuture<?> nettyResponseFuture, LongSupplier nanoClock) {
         if (requestTimeoutInMs <= -1) {
             return Long.MAX_VALUE;
         }
-        long spent = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - nettyResponseFuture.getStartNanos());
+        long spent = TimeUnit.NANOSECONDS.toMillis(nanoClock.getAsLong() - nettyResponseFuture.getStartNanos());
         return requestTimeoutInMs - spent;
     }
 
@@ -207,7 +224,7 @@ public void startReadTimeout() {
 
     void startReadTimeout(@Nullable ReadTimeoutTimerTask task) {
         if (requestTimeoutTask == null
-                || !requestTimeoutTask.isClaimed() && readTimeoutValue < requestTimeoutMillisTime - unpreciseMillisTime()) {
+                || !requestTimeoutTask.isClaimed() && readTimeoutValue < requestTimeoutMillisTime - millisClock.getAsLong()) {
             // only schedule a new readTimeout if the requestTimeout doesn't happen first
             if (task == null) {
                 // first call triggered from outside (else is read timeout is re-scheduling itself)
@@ -239,7 +256,7 @@ private static void release(@Nullable TimeoutTimerTask task) {
     private long remainingRequestTimeout() {
         // 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);
+        return Math.max(requestTimeoutMillisTime - millisClock.getAsLong(), 0L);
     }
 
     /**