diff --git a/client/src/jmh/java/org/asynchttpclient/bench/RequestTimeoutsArmBenchmark.java b/client/src/jmh/java/org/asynchttpclient/bench/RequestTimeoutsArmBenchmark.java
new file mode 100644
index 000000000..5c2948252
--- /dev/null
+++ b/client/src/jmh/java/org/asynchttpclient/bench/RequestTimeoutsArmBenchmark.java
@@ -0,0 +1,133 @@
+/*
+ *    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.bench;
+
+import io.netty.util.HashedWheelTimer;
+import io.netty.util.Timeout;
+import io.netty.util.TimerTask;
+import org.asynchttpclient.AsyncCompletionHandler;
+import org.asynchttpclient.AsyncHttpClientConfig;
+import org.asynchttpclient.DefaultAsyncHttpClient;
+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 org.asynchttpclient.netty.request.NettyRequestSender;
+import org.asynchttpclient.netty.timeout.TimeoutsHolder;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.TearDown;
+
+import java.lang.reflect.Field;
+import java.net.InetSocketAddress;
+import java.time.Duration;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * What arming and cancelling an exchange's timeouts costs on the client's {@link HashedWheelTimer}: one
+ * {@link TimeoutsHolder} built, started, given its read timeout and cancelled, as every attempt does.
+ * <p>
+ * {@code readTimeoutMs} picks which timeout comes first: at 60000 the read timeout is not shorter than what is
+ * left of the 60 s request timeout, so only the request timeout needs arming; at 10000 the read timeout is
+ * due first and has to be served as well. {@code rawTimerEntry} is one bare {@code newTimeout} plus
+ * {@code cancel}, the floor for a single entry.
+ * <p>
+ * The timer ticks every millisecond rather than the default 100: its worker moves at most 100000 new entries
+ * into the wheel per tick, so at the rates measured here a 100 ms tick leaves a backlog that grows for the
+ * whole run and turns the result into a measure of the collector.
+ * <p>
+ * Run with several thread counts, since the timer's queues and pending counter are shared by every caller:
+ * {@code -t 1}, {@code -t 8}, {@code -t 32}, with {@code -prof gc}.
+ */
+@State(Scope.Benchmark)
+@BenchmarkMode(Mode.Throughput)
+@OutputTimeUnit(TimeUnit.MICROSECONDS)
+public class RequestTimeoutsArmBenchmark {
+
+    private static final TimerTask NOOP = timeout -> {
+    };
+
+    @Param({"60000", "10000"})
+    public int readTimeoutMs;
+
+    private HashedWheelTimer timer;
+    private DefaultAsyncHttpClient client;
+    private AsyncHttpClientConfig config;
+    private NettyRequestSender requestSender;
+    private InetSocketAddress remoteAddress;
+
+    @State(Scope.Thread)
+    public static class Exchange {
+        NettyResponseFuture<?> future;
+
+        @Setup
+        public void setup() {
+            Request request = new RequestBuilder().setUrl("http://localhost:8080/").build();
+            future = new NettyResponseFuture<>(request, new AsyncCompletionHandler<Object>() {
+                @Override
+                public Object onCompleted(Response response) {
+                    return null;
+                }
+            }, null, 0, ChannelPoolPartitioning.PerHostChannelPoolPartitioning.INSTANCE, null, null);
+        }
+    }
+
+    @Setup
+    public void setup() throws ReflectiveOperationException {
+        timer = new HashedWheelTimer(1, TimeUnit.MILLISECONDS, 512);
+        timer.start();
+        config = new DefaultAsyncHttpClientConfig.Builder()
+                .setNettyTimer(timer)
+                .setRequestTimeout(Duration.ofMillis(60000))
+                .setReadTimeout(Duration.ofMillis(readTimeoutMs))
+                .build();
+        client = new DefaultAsyncHttpClient(config);
+        Field field = DefaultAsyncHttpClient.class.getDeclaredField("requestSender");
+        field.setAccessible(true);
+        requestSender = (NettyRequestSender) field.get(client);
+        remoteAddress = InetSocketAddress.createUnresolved("localhost", 8080);
+    }
+
+    @TearDown
+    public void tearDown() throws Exception {
+        client.close();
+        timer.stop();
+    }
+
+    @Benchmark
+    public TimeoutsHolder armAndCancel(Exchange exchange) {
+        TimeoutsHolder holder = new TimeoutsHolder(timer, null, exchange.future, requestSender, config, remoteAddress);
+        holder.start();
+        holder.startReadTimeout();
+        holder.cancel();
+        return holder;
+    }
+
+    @Benchmark
+    public Timeout rawTimerEntry() {
+        Timeout timeout = timer.newTimeout(NOOP, 60000, TimeUnit.MILLISECONDS);
+        timeout.cancel();
+        return timeout;
+    }
+}
diff --git a/client/src/jmh/java/org/asynchttpclient/bench/RequestTimeoutsFireBenchmark.java b/client/src/jmh/java/org/asynchttpclient/bench/RequestTimeoutsFireBenchmark.java
new file mode 100644
index 000000000..1bfaff41f
--- /dev/null
+++ b/client/src/jmh/java/org/asynchttpclient/bench/RequestTimeoutsFireBenchmark.java
@@ -0,0 +1,175 @@
+/*
+ *    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.bench;
+
+import io.netty.util.HashedWheelTimer;
+import io.netty.util.Timeout;
+import io.netty.util.Timer;
+import io.netty.util.TimerTask;
+import org.asynchttpclient.AsyncCompletionHandler;
+import org.asynchttpclient.AsyncHttpClientConfig;
+import org.asynchttpclient.DefaultAsyncHttpClient;
+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 org.asynchttpclient.netty.request.NettyRequestSender;
+import org.asynchttpclient.netty.timeout.TimeoutsHolder;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.TearDown;
+
+import java.lang.reflect.Field;
+import java.net.InetSocketAddress;
+import java.time.Duration;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * An exchange that outlives its read window: the {@link RequestTimeoutsArmBenchmark#armAndCancel} lifecycle
+ * with the entry that checks the read timeout firing {@code fires} times in between, each time finding the
+ * channel read from within the window and arming again. Read timeout 10 s, request timeout 60 s, on the
+ * client's {@link HashedWheelTimer}.
+ * <p>
+ * The benchmark fires the entry itself, on its own thread, rather than waiting out the window: it takes the
+ * shortest entry the attempt has pending out of the wheel and runs its task. Taking it out is a cancel, which
+ * a real expiry does not cost the caller; both builds pay it alike, once per fire.
+ * <p>
+ * Run with {@code -t 1}, {@code -t 8}, {@code -t 32}, with {@code -prof gc}.
+ */
+@State(Scope.Benchmark)
+@BenchmarkMode(Mode.Throughput)
+@OutputTimeUnit(TimeUnit.MICROSECONDS)
+public class RequestTimeoutsFireBenchmark {
+
+    @Param({"1", "5"})
+    public int fires;
+
+    private HashedWheelTimer timer;
+    private DefaultAsyncHttpClient client;
+    private AsyncHttpClientConfig config;
+    private NettyRequestSender requestSender;
+    private InetSocketAddress remoteAddress;
+
+    @State(Scope.Thread)
+    public static class Exchange {
+        NettyResponseFuture<?> future;
+        ShortestEntryTimer timer;
+
+        @Setup
+        public void setup(RequestTimeoutsFireBenchmark benchmark) {
+            Request request = new RequestBuilder().setUrl("http://localhost:8080/").build();
+            future = new NettyResponseFuture<>(request, new AsyncCompletionHandler<Object>() {
+                @Override
+                public Object onCompleted(Response response) {
+                    return null;
+                }
+            }, null, 0, ChannelPoolPartitioning.PerHostChannelPoolPartitioning.INSTANCE, null, null);
+            timer = new ShortestEntryTimer(benchmark.timer);
+        }
+    }
+
+    /**
+     * Arms on the shared timer and remembers this thread's shortest pending entry, which is the one that checks
+     * the read timeout.
+     */
+    static final class ShortestEntryTimer implements Timer {
+        private final Timer delegate;
+        private TimerTask task;
+        private Timeout timeout;
+        private long delay = Long.MAX_VALUE;
+
+        ShortestEntryTimer(Timer delegate) {
+            this.delegate = delegate;
+        }
+
+        @Override
+        public Timeout newTimeout(TimerTask task, long delay, TimeUnit unit) {
+            Timeout timeout = delegate.newTimeout(task, delay, unit);
+            if (delay < this.delay) {
+                this.task = task;
+                this.timeout = timeout;
+                this.delay = delay;
+            }
+            return timeout;
+        }
+
+        void fireShortest() throws Exception {
+            TimerTask task = this.task;
+            Timeout timeout = this.timeout;
+            this.task = null;
+            this.timeout = null;
+            delay = Long.MAX_VALUE;
+            timeout.cancel();
+            task.run(timeout);
+        }
+
+        void reset() {
+            task = null;
+            timeout = null;
+            delay = Long.MAX_VALUE;
+        }
+
+        @Override
+        public Set<Timeout> stop() {
+            throw new UnsupportedOperationException();
+        }
+    }
+
+    @Setup
+    public void setup() throws ReflectiveOperationException {
+        timer = new HashedWheelTimer(1, TimeUnit.MILLISECONDS, 512);
+        timer.start();
+        config = new DefaultAsyncHttpClientConfig.Builder()
+                .setNettyTimer(timer)
+                .setRequestTimeout(Duration.ofMillis(60000))
+                .setReadTimeout(Duration.ofMillis(10000))
+                .build();
+        client = new DefaultAsyncHttpClient(config);
+        Field field = DefaultAsyncHttpClient.class.getDeclaredField("requestSender");
+        field.setAccessible(true);
+        requestSender = (NettyRequestSender) field.get(client);
+        remoteAddress = InetSocketAddress.createUnresolved("localhost", 8080);
+    }
+
+    @TearDown
+    public void tearDown() throws Exception {
+        client.close();
+        timer.stop();
+    }
+
+    @Benchmark
+    public TimeoutsHolder armFireAndCancel(Exchange exchange) throws Exception {
+        TimeoutsHolder holder = new TimeoutsHolder(exchange.timer, null, exchange.future, requestSender, config, remoteAddress);
+        holder.start();
+        holder.startReadTimeout();
+        for (int i = 0; i < fires; i++) {
+            exchange.future.touch();
+            exchange.timer.fireShortest();
+        }
+        holder.cancel();
+        exchange.timer.reset();
+        return holder;
+    }
+}
diff --git a/client/src/main/java/org/asynchttpclient/netty/timeout/ReadTimeoutTimerTask.java b/client/src/main/java/org/asynchttpclient/netty/timeout/ReadTimeoutTimerTask.java
index 1ee82da2a..01aedabee 100755
--- a/client/src/main/java/org/asynchttpclient/netty/timeout/ReadTimeoutTimerTask.java
+++ b/client/src/main/java/org/asynchttpclient/netty/timeout/ReadTimeoutTimerTask.java
@@ -23,7 +23,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 
 import static org.asynchttpclient.util.DateUtils.unpreciseMillisTime;
 
@@ -31,8 +31,11 @@ public class ReadTimeoutTimerTask extends TimeoutTimerTask implements Runnable {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(ReadTimeoutTimerTask.class);
 
+    private static final AtomicIntegerFieldUpdater<ReadTimeoutTimerTask> INDEFINITE_SUSPENSION_WARNING_LOGGED =
+            AtomicIntegerFieldUpdater.newUpdater(ReadTimeoutTimerTask.class, "indefiniteSuspensionWarningLogged");
+
     private final long readTimeout;
-    private final AtomicBoolean indefiniteSuspensionWarningLogged = new AtomicBoolean();
+    private volatile int indefiniteSuspensionWarningLogged;
 
     ReadTimeoutTimerTask(NettyResponseFuture<?> nettyResponseFuture, NettyRequestSender requestSender, TimeoutsHolder timeoutsHolder, long readTimeout) {
         super(nettyResponseFuture, requestSender, timeoutsHolder);
@@ -89,7 +92,7 @@ public void run(Timeout timeout) {
 
     void warnIfIndefinitelySuspended() {
         if (timeoutsHolder.isRequestTimeoutDisabled()
-                && indefiniteSuspensionWarningLogged.compareAndSet(false, true)) {
+                && INDEFINITE_SUSPENSION_WARNING_LOGGED.compareAndSet(this, 0, 1)) {
             LOGGER.warn("Response body reads for {} remain suspended while the request timeout is disabled; "
                             + "the exchange retains its transport resources until it is resumed or canceled",
                     nettyResponseFuture.getUri().toUrlWithoutUserInfo());
diff --git a/client/src/main/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTask.java b/client/src/main/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTask.java
index 1b913ea7a..80aaba7a0 100755
--- a/client/src/main/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTask.java
+++ b/client/src/main/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTask.java
@@ -16,6 +16,7 @@
 package org.asynchttpclient.netty.timeout;
 
 import io.netty.util.Timeout;
+import org.asynchttpclient.netty.NettyResponseBodyControl;
 import org.asynchttpclient.netty.NettyResponseFuture;
 import org.asynchttpclient.netty.request.NettyRequestSender;
 import org.asynchttpclient.util.StringBuilderPool;
@@ -49,6 +50,10 @@ public void run(Timeout timeout) {
             return;
         }
 
+        if (timeoutsHolder.isRequestTimeoutArmedEarly() && checkReadTimeout()) {
+            return;
+        }
+
         // in any case, cancel possible readTimeout sibling
         timeoutsHolder.cancel();
 
@@ -62,4 +67,56 @@ public void run(Timeout timeout) {
         long age = unpreciseMillisTime() - nettyResponseFuture.getStart();
         expire(message, age);
     }
+
+    /**
+     * Runs when this entry was armed ahead of the request deadline, for when the read timeout could be due, so
+     * that one entry serves both. Fails the exchange with a read timeout when it is due, and otherwise arms
+     * the entry again for whichever of the two comes next.
+     *
+     * @return {@code false} when the request deadline has passed as well, which leaves the request timeout to
+     *         be delivered as though the entry had been armed for it
+     */
+    private boolean checkReadTimeout() {
+        if (nettyResponseFuture.isDone()) {
+            timeoutsHolder.cancel();
+            return true;
+        }
+
+        long untilDeadline = timeoutsHolder.untilRequestDeadline();
+        if (untilDeadline == 0) {
+            return false;
+        }
+        long now = timeoutsHolder.millisTime();
+        boolean suspended = NettyResponseBodyControl.isSuspended(nettyResponseFuture);
+        if (timeoutsHolder.isReadTimeoutCovered() && !suspended) {
+            long readTimeout = timeoutsHolder.readTimeout();
+            long sinceLastTouch = now - nettyResponseFuture.getLastTouch();
+            if (sinceLastTouch >= readTimeout) {
+                StringBuilder sb = StringBuilderPool.DEFAULT.stringBuilder().append("Read timeout to ");
+                appendRemoteAddress(sb);
+                String message = sb.append(" after ").append(readTimeout).append(" ms").toString();
+                expire(message, sinceLastTouch);
+                timeoutsHolder.cancel();
+                return true;
+            }
+        }
+
+        // Read before release: a concurrent cancel cleans the field once the task is unclaimed.
+        NettyResponseFuture<?> future = nettyResponseFuture;
+        // Released before re-arming: the holder may hand the read timeout an entry of its own, which is only
+        // armed while this task is unclaimed.
+        done.set(false);
+        try {
+            timeoutsHolder.rearmRequestTimeout(this, now, untilDeadline, suspended);
+        } catch (RuntimeException e) {
+            // A timer that rejects the entry, being full or stopped, would leave the exchange with no deadline
+            // at all, so it is failed instead of left to hang. Only while this holder is still the exchange's:
+            // one that was retired meanwhile no longer speaks for it, and the channel may already be serving
+            // the next attempt or, back in the pool, another exchange.
+            if (timeoutsHolder.retire() && future.getTimeoutsHolder() == timeoutsHolder && !future.isDone()) {
+                requestSender.abort(future.channel(), future, e);
+            }
+        }
+        return true;
+    }
 }
diff --git a/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutTimerTask.java b/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutTimerTask.java
index 6e81b7007..d02b8e10d 100755
--- a/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutTimerTask.java
+++ b/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutTimerTask.java
@@ -28,6 +28,8 @@
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
 
 /**
  * A timeout that can be armed either on a {@link io.netty.util.Timer} or on an event loop; which one an
@@ -39,16 +41,23 @@
 public abstract class TimeoutTimerTask implements TimerTask {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(TimeoutTimerTask.class);
+    private static final AtomicReferenceFieldUpdater<TimeoutTimerTask, Object> HANDLE =
+            AtomicReferenceFieldUpdater.newUpdater(TimeoutTimerTask.class, Object.class, "handle");
+    private static final AtomicIntegerFieldUpdater<TimeoutTimerTask> ARMINGS =
+            AtomicIntegerFieldUpdater.newUpdater(TimeoutTimerTask.class, "armings");
 
     protected final AtomicBoolean done = new AtomicBoolean();
     protected final NettyRequestSender requestSender;
     final TimeoutsHolder timeoutsHolder;
     volatile NettyResponseFuture<?> nettyResponseFuture;
-    // The scheduled entry this task is armed on, one field per scheduler so that a scheduler changing its
-    // return type is a compile error rather than a cancellation that silently stops working. At most one is
-    // ever set. Held here rather than in a wrapper so arming allocates nothing beyond what the scheduler needs.
-    private volatile @Nullable Timeout timerHandle;
-    private volatile @Nullable ScheduledFuture<?> loopHandle;
+    // The scheduled entry this task is armed on: a timer's Timeout or an event loop's ScheduledFuture, which
+    // only the typed armedOn overloads below can store. One field, so that recording an entry and taking it
+    // back are each a single atomic step: a task that re-arms itself can record an entry from the thread it
+    // fires on while the thread that armed it is still recording the previous one. Held here rather than in a
+    // wrapper so arming allocates nothing beyond what the scheduler needs.
+    private volatile @Nullable Object handle;
+    // How many times this task has been armed, so that an entry recorded late can tell it has been superseded.
+    private volatile int armings;
 
     TimeoutTimerTask(NettyResponseFuture<?> nettyResponseFuture, NettyRequestSender requestSender, TimeoutsHolder timeoutsHolder) {
         this.nettyResponseFuture = nettyResponseFuture;
@@ -56,16 +65,38 @@ public abstract class TimeoutTimerTask implements TimerTask {
         this.timeoutsHolder = timeoutsHolder;
     }
 
-    void armedOn(Timeout handle) {
-        // Each clears the other, so a handle left over from a previous arming cannot mask the live one and
-        // leave its entry sitting in a scheduler, holding the future until a deadline nobody is waiting for.
-        loopHandle = null;
-        timerHandle = handle;
+    /**
+     * Starts arming this task. The value returned identifies the arming to {@code armedOn}.
+     */
+    int beginArming() {
+        return ARMINGS.incrementAndGet(this);
+    }
+
+    void armedOn(Timeout handle, int arming) {
+        record(handle, arming);
+    }
+
+    void armedOn(ScheduledFuture<?> handle, int arming) {
+        record(handle, arming);
     }
 
-    void armedOn(ScheduledFuture<?> handle) {
-        timerHandle = null;
-        loopHandle = handle;
+    /**
+     * Records {@code entry}, unless the task has been armed again since {@code arming} began. That is an arming
+     * that lost the race against its own firing: its run re-armed the task, and overwriting the newer entry
+     * would leave it in its scheduler with nobody to cancel it, holding the future until a deadline nobody is
+     * waiting for. The entry passed over has already run, so there is nothing left to cancel on it.
+     */
+    private void record(Object entry, int arming) {
+        for (;;) {
+            Object current = handle;
+            if (armings != arming) {
+                return;
+            }
+            // Fails when a newer arming recorded its entry after the check above, which the next pass then sees.
+            if (HANDLE.compareAndSet(this, current, entry)) {
+                return;
+            }
+        }
     }
 
     /**
@@ -75,14 +106,14 @@ void armedOn(ScheduledFuture<?> handle) {
      * @return whether an entry was taken back out of its scheduler before it could run
      */
     boolean cancelArmed() {
-        Timeout timer = timerHandle;
-        if (timer != null) {
-            timerHandle = null;
-            return timer.cancel();
+        // Taken in one step: a read followed by a separate clear could clear an entry the task recorded in
+        // between from its own firing, and that one would then never be cancelled.
+        Object entry = HANDLE.getAndSet(this, null);
+        if (entry instanceof Timeout) {
+            return ((Timeout) entry).cancel();
         }
-        ScheduledFuture<?> scheduled = loopHandle;
-        if (scheduled != null) {
-            loopHandle = null;
+        if (entry != null) {
+            ScheduledFuture<?> scheduled = (ScheduledFuture<?>) entry;
             try {
                 return scheduled.cancel(false);
             } catch (RejectedExecutionException e) {
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 8c5c269d6..8b613ab00 100755
--- a/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java
+++ b/client/src/main/java/org/asynchttpclient/netty/timeout/TimeoutsHolder.java
@@ -30,7 +30,7 @@
 import java.net.InetSocketAddress;
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 import java.util.function.LongSupplier;
 
 /**
@@ -42,12 +42,34 @@ public class TimeoutsHolder {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(TimeoutsHolder.class);
 
-    private final AtomicBoolean cancelled = new AtomicBoolean();
-    private final AtomicBoolean started = new AtomicBoolean();
+    private static final AtomicIntegerFieldUpdater<TimeoutsHolder> CANCELLED =
+            AtomicIntegerFieldUpdater.newUpdater(TimeoutsHolder.class, "cancelled");
+    private static final AtomicIntegerFieldUpdater<TimeoutsHolder> STARTED =
+            AtomicIntegerFieldUpdater.newUpdater(TimeoutsHolder.class, "started");
+    private static final AtomicIntegerFieldUpdater<TimeoutsHolder> READ_STATE =
+            AtomicIntegerFieldUpdater.newUpdater(TimeoutsHolder.class, "readState");
+
+    // How the read timeout is armed. Separate: on an entry of its own. Pending: start armed the request
+    // timeout's entry for when a read timeout would be due, so that once the read timeout starts that one
+    // entry serves both. Covered: the read timeout has started and rides on the request timeout's entry.
+    private static final int READ_SEPARATE = 0;
+    private static final int READ_PENDING = 1;
+    private static final int READ_COVERED = 2;
+
+    private volatile int cancelled;
+    private volatile int started;
+    private volatile int readState;
+    // Whether the request timeout's entry is armed ahead of the request deadline, to check the read timeout.
+    private volatile boolean requestTimeoutArmedEarly;
     private final Timer nettyTimer;
     private volatile @Nullable EventExecutor eventExecutor;
     private final NettyRequestSender requestSender;
     private final long requestTimeoutMillisTime;
+    // The same deadline on nanoClock, for everything that re-arms or checks it after start: a timer entry is
+    // already timed on a monotonic clock, and a deadline re-derived from the wall clock would move with every
+    // step of it. Set only where something will read it -- an entry armed early, or one rehomeOn may move --
+    // since reading the clock is a measurable part of what an attempt that needs neither costs.
+    private volatile long requestDeadlineNanos;
     private final long requestTimeoutValue;
     private final boolean absoluteDeadline;
     private final long readTimeoutValue;
@@ -106,8 +128,14 @@ 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 = millisClock.getAsLong()
-                    + (absoluteDeadline ? remainingBudget(requestTimeoutInMs, nettyResponseFuture, nanoClock) : requestTimeoutInMs);
+            long untilDeadline = absoluteDeadline
+                    ? remainingBudget(requestTimeoutInMs, nettyResponseFuture, nanoClock) : requestTimeoutInMs;
+            requestTimeoutMillisTime = millisClock.getAsLong() + untilDeadline;
+            if (absoluteDeadline) {
+                // Straight off the exchange's start rather than off another clock reading, so that time passing
+                // between the two cannot extend it.
+                requestDeadlineNanos = nettyResponseFuture.getStartNanos() + TimeUnit.MILLISECONDS.toNanos(requestTimeoutInMs);
+            }
             requestTimeoutTask = new RequestTimeoutTimerTask(nettyResponseFuture, requestSender, this, requestTimeoutInMs);
         } else {
             requestTimeoutMillisTime = -1L;
@@ -129,7 +157,7 @@ public TimeoutsHolder(Timer nettyTimer, @Nullable EventExecutor eventExecutor, N
      * cancel to sit in its scheduler until the full deadline, pinning the task, the future and the channel.
      */
     public void start() {
-        if (!started.compareAndSet(false, true)) {
+        if (!STARTED.compareAndSet(this, 0, 1)) {
             return;
         }
         if (requestTimeoutTask != null) {
@@ -137,8 +165,23 @@ public void start() {
             // 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, nanoClock), 0L) : requestTimeoutValue);
+            long untilDeadline = absoluteDeadline
+                    ? Math.max(remainingBudget(requestTimeoutValue, nettyResponseFuture, nanoClock), 0L) : requestTimeoutValue;
+            boolean early = readTimeoutValue != -1 && readTimeoutValue < untilDeadline;
+            if (!absoluteDeadline && (early || useEventLoopTimeouts)) {
+                // Anchored where the entry is armed, from the same duration.
+                requestDeadlineNanos = nanoClock.getAsLong() + TimeUnit.MILLISECONDS.toNanos(requestTimeoutValue);
+            }
+            if (early) {
+                // The read timeout starts once the request is written, and would then take an entry of its own
+                // that the exchange almost always cancels unfired. Arming this one for when a read timeout
+                // would be due instead lets it serve both: every entry costs a shared timer two queue
+                // operations and a slot on its single thread, and an event loop a wakeup.
+                readState = READ_PENDING;
+                armRequestTimeout(requestTimeoutTask, readTimeoutValue, true);
+            } else {
+                armRequestTimeout(requestTimeoutTask, untilDeadline, false);
+            }
         }
     }
 
@@ -186,10 +229,42 @@ long requestTimeoutMillisTime() {
         return requestTimeoutMillisTime;
     }
 
+    // The clock lastTouch is read against, which tests replace.
+    long millisTime() {
+        return millisClock.getAsLong();
+    }
+
     boolean isRequestTimeoutDisabled() {
         return requestTimeoutTask == null;
     }
 
+    long readTimeout() {
+        return readTimeoutValue;
+    }
+
+    boolean isRequestTimeoutArmedEarly() {
+        return requestTimeoutArmedEarly;
+    }
+
+    boolean isReadTimeoutCovered() {
+        return readState == READ_COVERED;
+    }
+
+    /**
+     * What is left until the request deadline in whole milliseconds, rounded up so an entry never runs ahead of
+     * it, and floored at zero: a scheduler has no use for a negative delay, and the task has to run either way,
+     * since running is what fails the exchange. Zero exactly when the deadline has passed. One clock reading
+     * answers both whether it has and how long to re-arm for, so an entry that fires early reads it once.
+     */
+    long untilRequestDeadline() {
+        long left = requestDeadlineNanos - nanoClock.getAsLong();
+        if (left <= 0) {
+            return 0L;
+        }
+        long millis = TimeUnit.NANOSECONDS.toMillis(left);
+        return TimeUnit.MILLISECONDS.toNanos(millis) == left ? millis : millis + 1;
+    }
+
     /**
      * 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
@@ -202,10 +277,10 @@ public void rehomeOn(EventExecutor executor) {
         }
         eventExecutor = executor;
         RequestTimeoutTimerTask task = requestTimeoutTask;
-        if (task == null || cancelled.get() || task.isClaimed() || !task.cancelArmed()) {
+        if (task == null || cancelled != 0 || task.isClaimed() || !task.cancelArmed()) {
             return;
         }
-        arm(task, remainingRequestTimeout());
+        armRequestTimeoutFrom(task, millisClock.getAsLong(), untilRequestDeadline(), false);
     }
 
     public void setResolvedRemoteAddress(InetSocketAddress address) {
@@ -217,11 +292,65 @@ InetSocketAddress remoteAddress() {
     }
 
     public void startReadTimeout() {
-        if (readTimeoutValue != -1) {
+        if (readTimeoutValue != -1 && !coverReadTimeout()) {
             startReadTimeout(null);
         }
     }
 
+    /**
+     * Hands the read timeout to the request timeout's entry when {@link #start} armed that one early for it.
+     * Loses to an entry that already fired without finding a read timeout to check, which hands the read
+     * timeout back to an entry of its own.
+     */
+    private boolean coverReadTimeout() {
+        int state = readState;
+        return state == READ_COVERED || state == READ_PENDING && READ_STATE.compareAndSet(this, READ_PENDING, READ_COVERED);
+    }
+
+    /**
+     * Re-arms the request timeout after its entry fired ahead of the request deadline and found no timeout due.
+     *
+     * @param untilDeadline what {@link #untilRequestDeadline} returned when the entry fired
+     * @param suspended     whether response body reads are suspended, which holds off the read timeout for a
+     *                      whole window however long ago the channel was last read from
+     */
+    void rearmRequestTimeout(RequestTimeoutTimerTask task, long now, long untilDeadline, boolean suspended) {
+        // A read timeout that has not started by now may not start for a long time -- a slow connect or upload
+        // -- and checking for it every window would wake the scheduler for nothing, so it gets its own entry
+        // when it does start. Read first: on most fires the read timeout is long covered, and a failed CAS
+        // costs as much as one that succeeds.
+        if (readState == READ_PENDING) {
+            READ_STATE.compareAndSet(this, READ_PENDING, READ_SEPARATE);
+        }
+        armRequestTimeoutFrom(task, now, untilDeadline, suspended);
+    }
+
+    private void armRequestTimeoutFrom(RequestTimeoutTimerTask task, long now, long untilDeadline, boolean suspended) {
+        long untilRead = Long.MAX_VALUE;
+        int state = readState;
+        if (state == READ_PENDING) {
+            untilRead = readTimeoutValue;
+        } else if (state == READ_COVERED) {
+            // Capped at a whole window, so a wall clock stepping back cannot put the next check out of reach.
+            untilRead = suspended ? readTimeoutValue
+                    : Math.min(Math.max(nettyResponseFuture.getLastTouch() + readTimeoutValue - now, 0L), readTimeoutValue);
+        }
+        if (untilRead < untilDeadline) {
+            armRequestTimeout(task, untilRead, true);
+        } else {
+            armRequestTimeout(task, untilDeadline, false);
+        }
+    }
+
+    private void armRequestTimeout(RequestTimeoutTimerTask task, long delay, boolean early) {
+        // Written before arming: on an event loop the entry can run before arm returns. Only when it changes,
+        // since an entry that keeps re-arming early would otherwise pay a volatile write on every fire.
+        if (requestTimeoutArmedEarly != early) {
+            requestTimeoutArmedEarly = early;
+        }
+        arm(task, delay);
+    }
+
     void startReadTimeout(@Nullable ReadTimeoutTimerTask task) {
         if (requestTimeoutTask == null
                 || !requestTimeoutTask.isClaimed() && readTimeoutValue < requestTimeoutMillisTime - millisClock.getAsLong()) {
@@ -240,10 +369,21 @@ void startReadTimeout(@Nullable ReadTimeoutTimerTask task) {
     }
 
     public void cancel() {
-        if (cancelled.compareAndSet(false, true)) {
+        retire();
+    }
+
+    /**
+     * {@link #cancel}, telling whether this call is the one that cancelled, so that a caller acting on the
+     * exchange's behalf can tell a holder still in charge of it from one a completion, a redirect, a retry or
+     * an auth replay has already retired.
+     */
+    boolean retire() {
+        if (CANCELLED.compareAndSet(this, 0, 1)) {
             release(requestTimeoutTask);
             release(readTimeoutTask);
+            return true;
         }
+        return false;
     }
 
     private static void release(@Nullable TimeoutTimerTask task) {
@@ -253,12 +393,6 @@ 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 - millisClock.getAsLong(), 0L);
-    }
-
     /**
      * Arms {@code task} to run after {@code delay} milliseconds, recording the scheduled entry on the task so it
      * can cancel itself later. Leaves it unarmed when the client is shutting down, in which case there is no
@@ -274,6 +408,7 @@ private <T extends TimeoutTimerTask & Runnable> void arm(T task, long delay) {
         if (requestSender == null || requestSender.isClosed()) {
             return;
         }
+        int arming = task.beginArming();
         EventExecutor executor = eventExecutor;
         if (executor != null && !executor.isShuttingDown()) {
             ScheduledFuture<?> handle = null;
@@ -288,7 +423,7 @@ private <T extends TimeoutTimerTask & Runnable> void arm(T task, long delay) {
             // Outside the try: only the schedule above may fall back to the timer. Anything thrown while
             // recording or unwinding the entry belongs to an exchange that is already armed.
             if (handle != null) {
-                task.armedOn(handle);
+                task.armedOn(handle, arming);
                 cancelIfRaced(task);
                 return;
             }
@@ -296,7 +431,7 @@ private <T extends TimeoutTimerTask & Runnable> void arm(T task, long delay) {
         if (nettyTimer == null) {
             return;
         }
-        task.armedOn(nettyTimer.newTimeout(task, delay, TimeUnit.MILLISECONDS));
+        task.armedOn(nettyTimer.newTimeout(task, delay, TimeUnit.MILLISECONDS), arming);
         cancelIfRaced(task);
     }
 
@@ -306,7 +441,7 @@ private <T extends TimeoutTimerTask & Runnable> void arm(T task, long delay) {
      * would sit in the scheduler until the full deadline, waking a loop for a request that is long done.
      */
     private void cancelIfRaced(TimeoutTimerTask task) {
-        if (cancelled.get()) {
+        if (cancelled != 0) {
             release(task);
         }
     }
diff --git a/client/src/test/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTaskTest.java b/client/src/test/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTaskTest.java
new file mode 100644
index 000000000..17ad5e359
--- /dev/null
+++ b/client/src/test/java/org/asynchttpclient/netty/timeout/RequestTimeoutTimerTaskTest.java
@@ -0,0 +1,609 @@
+/*
+ *    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.netty.channel.embedded.EmbeddedChannel;
+import io.netty.util.HashedWheelTimer;
+import io.netty.util.Timeout;
+import io.netty.util.Timer;
+import io.netty.util.concurrent.EventExecutor;
+import io.netty.util.concurrent.ScheduledFuture;
+import org.asynchttpclient.AsyncCompletionHandler;
+import org.asynchttpclient.AsyncHttpClientConfig;
+import org.asynchttpclient.AsyncHttpClientState;
+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.NettyResponseBodyControl;
+import org.asynchttpclient.netty.NettyResponseFuture;
+import org.asynchttpclient.netty.channel.ChannelManager;
+import org.asynchttpclient.netty.request.NettyRequestSender;
+import org.jetbrains.annotations.Nullable;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.Field;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.function.Consumer;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * One timer entry serving both the request and the read timeout of an exchange. When the read timeout is the
+ * shorter of the two, {@link TimeoutsHolder#start} arms the request timeout's entry for when a read timeout
+ * would be due, and the read timeout then rides on it instead of taking an entry of its own.
+ * <p>
+ * The holder's clocks, the timer and the event loop are all fakes: the schedulers record what is armed and run
+ * nothing, so each test fires an entry by hand at the instant it sets on the clocks. The wall clock can be
+ * stepped apart from the monotonic one, as a clock correction does.
+ */
+public class RequestTimeoutTimerTaskTest {
+
+    private static final long READ_TIMEOUT_MS = 1_000;
+    private static final long REQUEST_TIMEOUT_MS = 5_000;
+    private static final long ONE_HOUR_MS = TimeUnit.HOURS.toMillis(1);
+
+    private final List<Armed> armed = new ArrayList<>();
+    private long elapsedMillis;
+    private long wallClockStepMillis;
+    private long wallOrigin;
+    private long nanoOrigin;
+    // When set, the first entry armed runs before the scheduler hands its handle back, as it can when the
+    // arming thread is descheduled for longer than the entry's delay.
+    private boolean firstEntryRunsBeforeItIsRecorded;
+    // When set, the monotonic clock jumps ahead by this much right after it is next read.
+    private long pauseAfterNextNanoReadMillis;
+    private int nanoReads;
+    // Runs as each entry is armed, before the scheduler hands its handle back.
+    private Consumer<Armed> onArm = entry -> { };
+
+    private HashedWheelTimer senderTimer;
+    private ChannelManager channelManager;
+    private Timer timer;
+    private EventExecutor loop;
+    private NettyResponseFuture<?> future;
+    private NettyRequestSender requestSender;
+    private AsyncHttpClientConfig config;
+
+    @BeforeEach
+    void setUp() {
+        timer = mock(Timer.class);
+        when(timer.newTimeout(any(), anyLong(), any())).thenAnswer(invocation -> {
+            Timeout timeout = mock(Timeout.class);
+            Armed entry = arm(invocation.getArgument(0), invocation.getArgument(1), timeout);
+            when(timeout.cancel()).thenAnswer(ignored -> !entry.ran);
+            return handBack(entry, timeout);
+        });
+        loop = mock(EventExecutor.class);
+        when(loop.schedule(any(Runnable.class), anyLong(), any())).thenAnswer(invocation -> {
+            ScheduledFuture<?> scheduled = mock(ScheduledFuture.class);
+            Armed entry = arm(invocation.getArgument(0), invocation.getArgument(1), scheduled);
+            when(scheduled.cancel(anyBoolean())).thenAnswer(ignored -> !entry.ran);
+            return handBack(entry, scheduled);
+        });
+    }
+
+    @AfterEach
+    void tearDown() {
+        if (channelManager != null) {
+            channelManager.close();
+        }
+        if (senderTimer != null) {
+            senderTimer.stop();
+        }
+    }
+
+    @Test
+    public void oneEntryServesBothTimeouts() {
+        TimeoutsHolder holder = startedHolder(REQUEST_TIMEOUT_MS);
+        holder.startReadTimeout();
+
+        assertEquals(1, armed.size(), "the read timeout should not have taken an entry of its own");
+        assertInstanceOf(RequestTimeoutTimerTask.class, armed.get(0).task);
+        assertEquals(READ_TIMEOUT_MS, armed.get(0).delay);
+    }
+
+    @Test
+    public void anEarlyEntryWaitsOnlyForWhatIsLeftOfTheReadWindow() {
+        TimeoutsHolder holder = startedHolder(REQUEST_TIMEOUT_MS);
+        holder.startReadTimeout();
+
+        fire(0, 600);
+
+        assertFalse(future.isDone());
+        assertEquals(2, armed.size());
+        assertSame(armed.get(0).task, armed.get(1).task);
+        assertEquals(400, armed.get(1).delay, "the entry should be due when the read window runs out");
+    }
+
+    @Test
+    public void aReadTimeoutIsDeliveredFromTheSharedEntry() {
+        TimeoutsHolder holder = startedHolder(REQUEST_TIMEOUT_MS);
+        holder.startReadTimeout();
+
+        fire(0, READ_TIMEOUT_MS);
+
+        assertTimedOut("Read timeout");
+        assertEquals(1, armed.size(), "an expired exchange should arm nothing more");
+    }
+
+    @Test
+    public void aReadTimeoutThatHasNotStartedGetsItsOwnEntry() {
+        TimeoutsHolder holder = startedHolder(REQUEST_TIMEOUT_MS);
+
+        // Still connecting or writing when the entry fires: nothing to check yet, and checking again every
+        // window until the read timeout starts would wake the timer for nothing.
+        fire(0, READ_TIMEOUT_MS);
+        assertEquals(2, armed.size());
+        assertEquals(REQUEST_TIMEOUT_MS - READ_TIMEOUT_MS, armed.get(1).delay,
+                "the entry should go back to the request deadline");
+
+        holder.startReadTimeout();
+        assertEquals(3, armed.size());
+        assertInstanceOf(ReadTimeoutTimerTask.class, armed.get(2).task);
+        assertEquals(READ_TIMEOUT_MS, armed.get(2).delay);
+    }
+
+    @Test
+    public void anAttemptWithNothingToReArmDoesNotReadTheMonotonicClock() {
+        // The default: a read timeout no shorter than the request timeout, on the timer.
+        TimeoutsHolder holder = startedHolder(READ_TIMEOUT_MS);
+        holder.startReadTimeout();
+        holder.cancel();
+
+        assertEquals(0, nanoReads, "only an entry that may be re-armed needs the request deadline on the nano clock");
+    }
+
+    @Test
+    public void anEarlyFireReadsTheMonotonicClockOnce() {
+        TimeoutsHolder holder = startedHolder(REQUEST_TIMEOUT_MS);
+        holder.startReadTimeout();
+        nanoReads = 0;
+
+        fire(0, READ_TIMEOUT_MS / 2);
+
+        assertEquals(2, armed.size());
+        assertEquals(1, nanoReads, "one reading should both check the request deadline and time the next entry");
+    }
+
+    @Test
+    public void aRequestDeadlineAheadOfTheReadTimeoutTakesTheEntryAlone() {
+        TimeoutsHolder holder = startedHolder(READ_TIMEOUT_MS / 2);
+        holder.startReadTimeout();
+
+        assertEquals(1, armed.size());
+        assertEquals(READ_TIMEOUT_MS / 2, armed.get(0).delay);
+
+        fire(0, READ_TIMEOUT_MS / 2);
+        assertTimedOut("Request timeout");
+    }
+
+    @Test
+    public void anAbsoluteDeadlineWithLessLeftThanTheReadTimeoutTakesTheEntryAlone() {
+        // An earlier hop already spent most of the exchange's budget.
+        TimeoutsHolder holder = holder(REQUEST_TIMEOUT_MS, true, false, null, REQUEST_TIMEOUT_MS - READ_TIMEOUT_MS / 2);
+        holder.start();
+        holder.startReadTimeout();
+
+        assertEquals(1, armed.size(), "the read timeout cannot be due before what is left of the request timeout");
+        assertEquals(READ_TIMEOUT_MS / 2, armed.get(0).delay);
+
+        fire(0, REQUEST_TIMEOUT_MS);
+        assertTimedOut("Request timeout");
+    }
+
+    @Test
+    public void anAbsoluteDeadlineIsNotExtendedByTimePassingWhileTheHolderIsBuilt() {
+        pauseAfterNextNanoReadMillis = 3_000;
+        TimeoutsHolder holder = holder(REQUEST_TIMEOUT_MS, true, false, null, 0);
+        holder.start();
+        assertEquals(1, armed.size());
+
+        // The read timeout has not started, so the entry goes back to the request deadline.
+        fire(0, 4_000);
+
+        assertEquals(2, armed.size());
+        assertEquals(REQUEST_TIMEOUT_MS - 4_000, armed.get(1).delay,
+                "the exchange's deadline should stay where its start and budget put it");
+    }
+
+    @Test
+    public void anEntryTheTimerRejectsFailsTheExchangeRatherThanLeaveItWithoutADeadline() {
+        TimeoutsHolder holder = holder(REQUEST_TIMEOUT_MS, false, false, null, 0);
+        future.setTimeoutsHolder(holder);
+        holder.startReadTimeout();
+        onArm = entry -> {
+            if (armed.size() == 2) {
+                throw new RejectedExecutionException("timer full");
+            }
+        };
+
+        fire(0, READ_TIMEOUT_MS / 2);
+
+        ExecutionException failure = assertThrows(ExecutionException.class, () -> future.get(0, TimeUnit.MILLISECONDS));
+        assertInstanceOf(RejectedExecutionException.class, failure.getCause());
+    }
+
+    @Test
+    public void aRejectedReArmOfARetiredHolderLeavesTheNextAttemptAlone() {
+        TimeoutsHolder holder = holder(REQUEST_TIMEOUT_MS, false, false, null, 0);
+        future.setTimeoutsHolder(holder);
+        holder.startReadTimeout();
+        // A redirect installs the next attempt's holder while the old entry is re-arming, and then the timer
+        // rejects the old entry.
+        onArm = entry -> {
+            if (armed.size() == 2) {
+                future.setTimeoutsHolder(new TimeoutsHolder(timer, null, future, requestSender, config, null,
+                        this::millis, this::nanos));
+                throw new RejectedExecutionException("timer full");
+            }
+        };
+
+        fire(0, READ_TIMEOUT_MS / 2);
+
+        assertFalse(future.isDone(), "the next attempt should not be failed by the holder it replaced");
+    }
+
+    @Test
+    public void aRejectedReArmOfAHolderTheFutureNoLongerHoldsAbortsNothing() {
+        TimeoutsHolder holder = holder(REQUEST_TIMEOUT_MS, false, false, null, 0);
+        future.setTimeoutsHolder(holder);
+        holder.startReadTimeout();
+        // Completion and the next attempt both take the holder off the future before they cancel it; the
+        // rejection lands in between.
+        onArm = entry -> {
+            if (armed.size() == 2) {
+                detachTimeoutsHolder();
+                throw new RejectedExecutionException("timer full");
+            }
+        };
+
+        fire(0, READ_TIMEOUT_MS / 2);
+
+        assertFalse(future.isDone(), "a holder the future no longer holds should leave the exchange alone");
+    }
+
+    @Test
+    public void aRejectedReArmAfterTheExchangeCompletedAbortsNothing() {
+        TimeoutsHolder holder = holder(REQUEST_TIMEOUT_MS, false, false, null, 0);
+        future.setTimeoutsHolder(holder);
+        holder.startReadTimeout();
+        onArm = entry -> {
+            if (armed.size() == 2) {
+                future.cancelTimeouts();
+                throw new RejectedExecutionException("timer full");
+            }
+        };
+
+        fire(0, READ_TIMEOUT_MS / 2);
+
+        assertFalse(future.isDone(), "a holder retired by completion should leave the exchange alone");
+    }
+
+    @Test
+    public void anEarlyEntryThatFiresPastTheRequestDeadlineDeliversTheRequestTimeout() {
+        TimeoutsHolder holder = startedHolder(REQUEST_TIMEOUT_MS);
+        holder.startReadTimeout();
+
+        // A timer runs an entry on the first tick at or after it is due, which can be past a deadline that
+        // fell between the two.
+        fire(0, REQUEST_TIMEOUT_MS);
+
+        assertTimedOut("Request timeout");
+    }
+
+    @Test
+    public void aWallClockStepForwardDoesNotBringTheRequestDeadlineForward() {
+        startedHolder(REQUEST_TIMEOUT_MS);
+
+        wallClockStepMillis = ONE_HOUR_MS;
+        fire(0, READ_TIMEOUT_MS);
+
+        assertFalse(future.isDone(), "a step of the wall clock is not time spent on the request");
+        assertEquals(2, armed.size());
+        assertEquals(REQUEST_TIMEOUT_MS - READ_TIMEOUT_MS, armed.get(1).delay);
+    }
+
+    @Test
+    public void aWallClockStepBackDoesNotPushTheRequestDeadlineBack() {
+        startedHolder(REQUEST_TIMEOUT_MS);
+
+        wallClockStepMillis = -ONE_HOUR_MS;
+        fire(0, READ_TIMEOUT_MS);
+
+        assertEquals(2, armed.size());
+        assertEquals(REQUEST_TIMEOUT_MS - READ_TIMEOUT_MS, armed.get(1).delay,
+                "the request deadline should stay where the monotonic clock put it");
+    }
+
+    @Test
+    public void aRequestTimeoutPastWhatTheMonotonicClockCanHoldIsNotTakenForPassed() {
+        TimeoutsHolder holder = holder(Long.MAX_VALUE, false, false, null, 0);
+        // Any positive reading plus the longest timeout overflows, which only a comparison by subtraction
+        // survives.
+        nanoOrigin = ONE_HOUR_MS;
+        holder.start();
+        holder.startReadTimeout();
+
+        fire(0, READ_TIMEOUT_MS / 2);
+
+        assertFalse(future.isDone(), "the request deadline is nowhere near");
+        assertEquals(2, armed.size());
+        assertTrue(holder.isRequestTimeoutArmedEarly());
+    }
+
+    @Test
+    public void aWallClockStepBackDoesNotPutTheNextReadCheckOutOfReach() {
+        TimeoutsHolder holder = startedHolder(REQUEST_TIMEOUT_MS);
+        holder.startReadTimeout();
+
+        wallClockStepMillis = -ONE_HOUR_MS;
+        fire(0, READ_TIMEOUT_MS / 2);
+
+        assertEquals(2, armed.size());
+        assertEquals(READ_TIMEOUT_MS, armed.get(1).delay, "the read timeout should be checked again within a window");
+        assertTrue(holder.isRequestTimeoutArmedEarly());
+    }
+
+    @Test
+    public void aReadTimeoutStartingWhileTheEntryReArmsGetsItsOwnEntry() {
+        TimeoutsHolder holder = startedHolder(REQUEST_TIMEOUT_MS);
+        // The request is written while the entry, finding no read timeout yet, is re-arming for the deadline.
+        onArm = entry -> {
+            if (armed.size() == 2) {
+                holder.startReadTimeout();
+            }
+        };
+
+        fire(0, READ_TIMEOUT_MS);
+
+        assertEquals(3, armed.size());
+        assertEquals(REQUEST_TIMEOUT_MS - READ_TIMEOUT_MS, armed.get(1).delay);
+        assertInstanceOf(ReadTimeoutTimerTask.class, armed.get(2).task, "the read timeout should not be lost");
+        assertEquals(READ_TIMEOUT_MS, armed.get(2).delay);
+
+        holder.cancel();
+        verify((Timeout) armed.get(1).timeout).cancel();
+        verify((Timeout) armed.get(2).timeout).cancel();
+    }
+
+    @Test
+    public void aSuspendedBodyHoldsOffTheReadTimeoutForAWholeWindow() {
+        TimeoutsHolder holder = startedHolder(REQUEST_TIMEOUT_MS);
+        holder.startReadTimeout();
+        EmbeddedChannel channel = new EmbeddedChannel();
+        try {
+            NettyResponseBodyControl.create(future, channel, () -> { }, ignored -> { }).suspend();
+
+            fire(0, 3 * READ_TIMEOUT_MS);
+
+            assertFalse(future.isDone(), "a suspended body is not a read timeout");
+            assertEquals(2, armed.size());
+            assertEquals(READ_TIMEOUT_MS, armed.get(1).delay);
+        } finally {
+            channel.finishAndReleaseAll();
+        }
+    }
+
+    @Test
+    public void cancellingAfterAnEarlyFireTakesTheNewEntryBackOut() {
+        TimeoutsHolder holder = startedHolder(REQUEST_TIMEOUT_MS);
+        holder.startReadTimeout();
+        fire(0, READ_TIMEOUT_MS / 2);
+
+        holder.cancel();
+
+        verify((Timeout) armed.get(1).timeout).cancel();
+        RequestTimeoutTimerTask task = (RequestTimeoutTimerTask) armed.get(1).task;
+        assertTrue(task.isClaimed());
+        assertNull(task.nettyResponseFuture, "a cancelled entry should not pin the exchange");
+    }
+
+    @Test
+    public void anEntryThatFiresBeforeItIsRecordedDoesNotHideItsSuccessorOnTheTimer() {
+        firstEntryRunsBeforeItIsRecorded = true;
+        TimeoutsHolder holder = startedHolder(REQUEST_TIMEOUT_MS);
+        assertEquals(2, armed.size(), "the first entry should have re-armed while it was still being recorded");
+
+        holder.cancel();
+
+        verify((Timeout) armed.get(1).timeout).cancel();
+    }
+
+    @Test
+    public void anEntryThatFiresBeforeItIsRecordedDoesNotHideItsSuccessorOnTheLoop() {
+        firstEntryRunsBeforeItIsRecorded = true;
+        TimeoutsHolder holder = holder(REQUEST_TIMEOUT_MS, false, true, loop, 0);
+        holder.start();
+        assertEquals(2, armed.size(), "the first entry should have re-armed while it was still being recorded");
+
+        holder.cancel();
+
+        verify((ScheduledFuture<?>) armed.get(1).timeout).cancel(false);
+    }
+
+    @Test
+    public void cancellingAfterAnEarlyFireOnTheLoopTakesTheNewEntryBackOut() {
+        TimeoutsHolder holder = holder(REQUEST_TIMEOUT_MS, false, true, loop, 0);
+        holder.start();
+        holder.startReadTimeout();
+
+        fire(0, READ_TIMEOUT_MS / 2);
+        assertEquals(2, armed.size());
+
+        holder.cancel();
+
+        verify((ScheduledFuture<?>) armed.get(1).timeout).cancel(false);
+    }
+
+    @Test
+    public void rehomingBeforeTheReadTimeoutStartsKeepsTheEntryEarly() {
+        // The connect path arms on the timer, before there is a channel whose loop could carry the entry.
+        TimeoutsHolder holder = holder(REQUEST_TIMEOUT_MS, false, true, null, 0);
+        holder.start();
+        assertEquals(1, armed.size());
+
+        elapsedMillis = 300;
+        holder.rehomeOn(loop);
+
+        verify((Timeout) armed.get(0).timeout).cancel();
+        assertEquals(2, armed.size());
+        assertInstanceOf(ScheduledFuture.class, armed.get(1).timeout, "the entry should have moved to the loop");
+        assertEquals(READ_TIMEOUT_MS, armed.get(1).delay, "the entry should still be armed to check a read timeout");
+        assertTrue(holder.isRequestTimeoutArmedEarly());
+    }
+
+    @Test
+    public void aCompletedExchangeArmsNothingMore() {
+        TimeoutsHolder holder = startedHolder(REQUEST_TIMEOUT_MS);
+        holder.startReadTimeout();
+        future.done();
+
+        // Inside the read window, where an exchange still running would be re-armed.
+        fire(0, READ_TIMEOUT_MS / 2);
+
+        assertEquals(1, armed.size(), "a finished exchange should arm nothing more");
+        assertTrue(((RequestTimeoutTimerTask) armed.get(0).task).isClaimed());
+    }
+
+    private TimeoutsHolder startedHolder(long requestTimeoutMs) {
+        TimeoutsHolder holder = holder(requestTimeoutMs, false, false, null, 0);
+        holder.start();
+        return holder;
+    }
+
+    /**
+     * @param spentMillis how long the exchange ran before this attempt, which only an absolute deadline counts
+     */
+    private TimeoutsHolder holder(long requestTimeoutMs, boolean absoluteDeadline, boolean useEventLoopTimeouts,
+                                  @Nullable EventExecutor executor, long spentMillis) {
+        config = config(requestTimeoutMs, useEventLoopTimeouts);
+        senderTimer = new HashedWheelTimer();
+        channelManager = new ChannelManager(config, senderTimer);
+        requestSender = new NettyRequestSender(config, channelManager, senderTimer, mock(AsyncHttpClientState.class));
+
+        Request request = new RequestBuilder().setUrl("http://example.com:12345").build();
+        future = new NettyResponseFuture<>(request, new AsyncCompletionHandler<Object>() {
+            @Override
+            public Object onCompleted(Response response) {
+                return null;
+            }
+        }, null, 0, ChannelPoolPartitioning.PerHostChannelPoolPartitioning.INSTANCE, null, null);
+        future.setUseAbsoluteRequestDeadline(absoluteDeadline);
+        wallOrigin = future.getLastTouch();
+        nanoOrigin = future.getStartNanos() + TimeUnit.MILLISECONDS.toNanos(spentMillis);
+        return new TimeoutsHolder(timer, executor, future, requestSender, config, null, this::millis, this::nanos);
+    }
+
+    // Takes the holder off the future without cancelling it, as NettyResponseFuture does before it cancels.
+    private void detachTimeoutsHolder() {
+        try {
+            Field field = NettyResponseFuture.class.getDeclaredField("timeoutsHolder");
+            field.setAccessible(true);
+            field.set(future, null);
+        } catch (ReflectiveOperationException e) {
+            throw new AssertionError(e);
+        }
+    }
+
+    private static AsyncHttpClientConfig config(long requestTimeoutMs, boolean useEventLoopTimeouts) {
+        return new DefaultAsyncHttpClientConfig.Builder()
+                .setReadTimeout(Duration.ofMillis(READ_TIMEOUT_MS))
+                .setRequestTimeout(Duration.ofMillis(requestTimeoutMs))
+                .setUseEventLoopTimeouts(useEventLoopTimeouts)
+                .build();
+    }
+
+    private Armed arm(Object task, long delay, Object handle) {
+        Armed entry = new Armed((TimeoutTimerTask) task, delay, handle);
+        armed.add(entry);
+        return entry;
+    }
+
+    private <H> H handBack(Armed entry, H handle) {
+        onArm.accept(entry);
+        if (firstEntryRunsBeforeItIsRecorded && armed.size() == 1) {
+            run(entry, READ_TIMEOUT_MS);
+        }
+        return handle;
+    }
+
+    private void fire(int entry, long atElapsedMillis) {
+        run(armed.get(entry), atElapsedMillis);
+    }
+
+    private void run(Armed entry, long atElapsedMillis) {
+        elapsedMillis = atElapsedMillis;
+        entry.ran = true;
+        try {
+            entry.task.run(entry.timeout instanceof Timeout ? (Timeout) entry.timeout : null);
+        } catch (Exception e) {
+            throw new AssertionError(e);
+        }
+    }
+
+    private void assertTimedOut(String prefix) {
+        ExecutionException failure = assertThrows(ExecutionException.class, () -> future.get(0, TimeUnit.MILLISECONDS));
+        assertInstanceOf(TimeoutException.class, failure.getCause());
+        assertTrue(failure.getCause().getMessage().startsWith(prefix), failure.getCause().getMessage());
+    }
+
+    private long millis() {
+        return wallOrigin + elapsedMillis + wallClockStepMillis;
+    }
+
+    private long nanos() {
+        nanoReads++;
+        long nanos = nanoOrigin + TimeUnit.MILLISECONDS.toNanos(elapsedMillis);
+        elapsedMillis += pauseAfterNextNanoReadMillis;
+        pauseAfterNextNanoReadMillis = 0;
+        return nanos;
+    }
+
+    private static final class Armed {
+        final TimeoutTimerTask task;
+        final long delay;
+        final Object timeout;
+        volatile boolean ran;
+
+        Armed(TimeoutTimerTask task, long delay, Object timeout) {
+            this.task = task;
+            this.delay = delay;
+            this.timeout = timeout;
+        }
+    }
+}