Arm request timeouts on an event loop (#2313)

## Problem

Request and read timeouts are armed on the client's `HashedWheelTimer`.
That has two
properties that only show up on short deadlines:

* **A wheel quantizes.** It fires on the first tick at or after the
deadline, so a
deadline near or below `hashedWheelTimerTickDuration` is rounded up to
it.
* **One thread carries every expiry** for the whole client, and
`HashedWheelTimer`'s
default `taskExecutor` is `ImmediateExecutor`, so each expiry runs
inline on the wheel
thread - including `future.completeExceptionally(...)` and therefore
whatever the caller
  chained onto the response future.

On a one-second budget the first costs 0.3% and nobody notices. On a
budget of tens of
milliseconds a tick is a large fraction of it, and a burst of expiries
has no headroom to
absorb before the wheel starts running late.

## Measured

2000 timeouts armed as one burst on Netty 4.2.16, JDK 17, tasks doing
nothing but
recording their own lag. This is the floor; real work on the firing
thread only adds to it.

| instrument | deadline | mean | p50 | p99 | max |
|---|---:|---:|---:|---:|---:|
| wheel, tick 5 ms | 20 ms | +2.7 | +2 | +5 | +5 |
| wheel, tick 1 ms | 20 ms | +1.3 | +1 | +2 | +2 |
| `EventLoop.schedule` | 20 ms | +0.0 | +0 | +0 | +0 |
| wheel, tick 5 ms | 1000 ms | +3.0 | +3 | +3 | +3 |
| wheel, tick 1 ms | 1000 ms | +1.9 | +2 | +2 | +2 |
| `EventLoop.schedule` | 1000 ms | +1.6 | +2 | +2 | +2 |

An event loop shows zero overshoot because it schedules by deadline and
derives its own
`select()` timeout from the nearest one. There is no quantum to round
to.

This was a throwaway probe rather than JMH - `client/src/jmh/java` is
not currently wired
into the build, so its benchmarks do not compile. Happy to add a proper
benchmark if that
is fixed first, or as part of this.

## Change

`AsyncHttpClientConfig#isUseEventLoopTimeouts()`, **off by default**,
arms the request and
read timeouts on an event loop instead of the timer.

* **Always the channel's own loop.** On the pooled path the channel is
already in hand, so
its loop is used and the timeout expires on the thread that would have
to close it. On the
connect path there is no channel yet - deliberately, so that the timeout
also bounds
address resolution and the connect - so it is armed on the timer and
moved onto the loop
once the connect succeeds. No other loop is ever used; see the review
round below for why
  that matters.
* **Arming allocates nothing extra.** The cancellation handle lives on
the task rather than
in a wrapper, and the existing `done` flag stands in for the scheduler's
already-expired
  flag, which the two schedulers spell differently.
* **Shutdown race closed.** `isShuttingDown()` can return false and
`schedule` reject
immediately after. Netty answers a rejected timeout with a logged
warning rather than an
exception, which would leave the exchange with nothing to end it, so a
rejection falls
  back to the timer.
* The connection-pool cleaner stays on the timer either way.

Off by default because the expiry - and so whatever the caller chained
onto the future -
then runs on an I/O thread, and blocking one stalls every connection it
serves. The javadoc
says so and points callers at `handleAsync`.

## Why not a wheel per event loop

That is how the Aerospike client solves the same problem:
`EventLoopBase` owns a
`HashedWheelTimer` that is a `Runnable` the loop ticks itself.
Deliberately not copied here.
A wheel arms in O(1) against O(log n) for a deadline queue, but at a few
thousand timeouts
per loop that is a dozen comparisons, while the quantization it
reintroduces costs
milliseconds on a 20 ms budget - the third row above is the whole point.
A wheel also has to
be ticked forever, waking every loop even with nothing armed. Aerospike
wrote its own because
its `EventLoop` abstracts over NIO, Netty and direct NIO and needed one
timer; AHC is
Netty-only and gets a per-loop deadline queue for free.

## Review round 1

Most of the substance of this PR changed in review, so the sections
above describe the
current shape rather than what was first pushed. Two things are worth
calling out here
because they were design errors, not polish:

**The loop is now only ever the channel's own.** It used to come from
`EventLoopGroup#next()`, which is almost never the loop the channel ends
up on:
`initAndRegister` draws from the same chooser, so the two agreed about
one time in N. Every
completion then cancelled an entry on a foreign loop, and until the
original deadline that
entry sat in a queue whose loop it would wake for a request that had
long finished. Drawing
from the chooser also shifted which loops connections land on. The
pooled path has the
channel in hand; the connect path arms on the timer, as before this
branch, and
`NettyConnectListener` moves the timeouts onto the loop once the connect
succeeds, next to
the `attachChannel` that publishes it on the future. That listener
already runs on the
channel's loop, so the move costs a same-thread schedule and no wakeup.

**Arming left the `TimeoutsHolder` constructor.** The task holds the
holder and can run the
moment it is armed, and an event loop does not round a short deadline up
to a tick, so the
expiry could reach a holder whose fields were not yet frozen, a future
that had not been
handed the holder, and on the pooled path a future with no channel
attached - which aborted
with `null` and left the pooled socket open. The caller now publishes
the holder, attaches
the channel, and calls `start()` last.

Also from the review: `arm` re-checks `cancelled` after recording its
handle, so an exchange
that finishes mid-arming cannot leave behind an entry nobody will
cancel; `cancelArmed`
catches the `RejectedExecutionException` that Netty's off-loop
cancellation path can raise on
a closing client, which had never escaped `ListenableFuture#cancel`
before; the cancellation
handle is two typed fields rather than an `Object` and `instanceof`;
`requestTimeoutArmed` is
gone in favour of a null test on the task; the rationale lives on
`isUseEventLoopTimeouts()` alone; and the new option sits in the `//
timeouts` group
everywhere rather than splitting the two `failedIpCooldown` entries.

## API compatibility

No `revapi` entries. `implements Runnable` and `run()` live on the two
subclasses rather than
on `TimeoutTimerTask`, which leaves that class's surface unchanged: both
subclasses already
declared `run(Timeout)` without a throws clause, so inside a subclass
`run()` calls its own
override and has nothing to catch. No dead handler, and nothing
narrowed.

The knock-on is that only the concrete classes are both a `TimerTask`
and a `Runnable`, so
`TimeoutsHolder#arm` takes that intersection as a type parameter.
Everything else is
additive: the existing `TimeoutsHolder` constructor is kept and
delegates, and nothing is
removed.

`start()` is called from `NettyResponseFuture#setTimeoutsHolder` rather
than by the sender.
`TimeoutsHolder` has a public constructor in an exported package, and
splitting the arming
out of it would otherwise leave an outside caller free to install a
holder and get an
exchange with no request timeout at all.

## Tests

Four cases in `EventLoopTimeoutTest`, asserting where an expiry is
delivered from rather than
what it does. They hand the config their own `Timer` and
`EventLoopGroup` so the assertions
are against those objects and not against thread names, which a pool
name containing `timer`
or a configured thread factory would have broken with no bug present:

- the timer default, by identity against the timer's own thread;
- a connecting exchange, on the loop of the channel
`onTcpConnectSuccess` reported;
- an exchange on a pooled channel, on the loop of the channel
`onConnectionPooled` reported,
which is also what says it reused the connection rather than opening one
of its own;
- a read timeout under the new mode, which is armed after the request is
written and so
  exercises a different arming path.

The group has eight loops. With two, a timeout armed on the wrong loop
is on the right one
half the time, and these assertions would have passed about half the
runs against the bug they
exist to catch; at eight, reverting `timeoutExecutor` to `next()` fails
the pooled case.

The connecting case, and the pooled case's first request, get a one
second budget on purpose.
A deadline reached before connecting would be delivered from the timer
quite correctly, there
being no channel to deliver it from, and would prove nothing either way;
and the pooled case's
first request is the cold one - class loading, the connect, the server's
first response - and
is meant to succeed.

One gap, called out rather than papered over: the
`RejectedExecutionException` fallback in
`arm` has no test. Reaching it needs a loop that answers
`isShuttingDown()` with `false` and
then rejects the schedule, and the executor comes from the channel, so
there is no way in
through the config. A test double for `EventExecutor` would do it if
that is acceptable.

## Verification

`mvnw clean verify` - BUILD SUCCESS, 1468 tests, 0 failures, 0 errors,
21 skipped. Error
Prone, NullAway clean; `revapi` clean with the one scoped entry above.

Caveat on the testing gate: `AGENTS.md` requires the build to run on JDK
11 and no JDK 11 is
installed on this machine, so it was run on **JDK 17** (also in the CI
matrix). The JDK 11
leg of CI on this PR is the real gate.

Claude Code on behalf of @pavel-ptashyts

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>

Author: pavel-ptashyts

client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java

@@ -114,6 +114,34 @@ public interface AsyncHttpClientConfig {
      */
     Duration getRequestTimeout();
 
+    /**
+     * Whether request and read timeouts are armed on an event loop rather than on {@link #getNettyTimer()}.
+     * <p>
+     * The timer is a hashed wheel: it fires on the first tick at or after the deadline, so a deadline near or
+     * below {@link #getHashedWheelTimerTickDuration()} is rounded up to it, and one thread carries every expiry
+     * for the whole client. An event loop instead schedules by deadline and derives its own select timeout from
+     * the nearest one, so nothing is rounded up, and the loops share the load rather than funnelling it through
+     * a single thread. Both effects matter most to short deadlines, where a tick is a large fraction of the
+     * budget and a burst of expiries has no headroom to absorb.
+     * <p>
+     * The cost is where the expiry runs. On the timer it runs on the timer thread; on an event loop it runs on
+     * an I/O thread, and so does whatever the caller chained onto the response future, because that future is
+     * completed from there. Blocking an I/O thread stalls every connection it serves, so a caller enabling this
+     * should hand its own work off with {@code handleAsync} or an {@code AsyncHandler} that does the same.
+     * That is why this is opt-in rather than the default.
+     * <p>
+     * The loop is always the one that owns the exchange's channel. Until there is a channel -- while an address
+     * is being resolved and a connection made -- the timer carries the timeout, and the exchange moves it onto
+     * the loop once the connection succeeds.
+     * <p>
+     * The connection-pool cleaner stays on the timer either way.
+     *
+     * @return {@code true} to arm request and read timeouts on an event loop
+     */
+    default boolean isUseEventLoopTimeouts() {
+        return false;
+    }
+
     /**
      * Is HTTP redirect enabled
      *

client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClientConfig.java

@@ -97,6 +97,7 @@
 import static org.asynchttpclient.config.AsyncHttpClientConfigDefaults.defaultStrict302Handling;
 import static org.asynchttpclient.config.AsyncHttpClientConfigDefaults.defaultTcpNoDelay;
 import static org.asynchttpclient.config.AsyncHttpClientConfigDefaults.defaultThreadPoolName;
+import static org.asynchttpclient.config.AsyncHttpClientConfigDefaults.defaultUseEventLoopTimeouts;
 import static org.asynchttpclient.config.AsyncHttpClientConfigDefaults.defaultUseInsecureTrustManager;
 import static org.asynchttpclient.config.AsyncHttpClientConfigDefaults.defaultUseLaxCookieEncoder;
 import static org.asynchttpclient.config.AsyncHttpClientConfigDefaults.defaultUseNativeTransport;
@@ -157,6 +158,7 @@ public class DefaultAsyncHttpClientConfig implements AsyncHttpClientConfig {
     private final Duration connectTimeout;
     private final Duration requestTimeout;
     private final Duration readTimeout;
+    private final boolean useEventLoopTimeouts;
     private final Duration shutdownQuietPeriod;
     private final Duration shutdownTimeout;
 
@@ -258,6 +260,7 @@ private DefaultAsyncHttpClientConfig(// http
                                          Duration connectTimeout,
                                          Duration requestTimeout,
                                          Duration readTimeout,
+                                         boolean useEventLoopTimeouts,
                                          Duration shutdownQuietPeriod,
                                          Duration shutdownTimeout,
 
@@ -367,6 +370,7 @@ private DefaultAsyncHttpClientConfig(// http
         this.connectTimeout = connectTimeout;
         this.requestTimeout = requestTimeout;
         this.readTimeout = readTimeout;
+        this.useEventLoopTimeouts = useEventLoopTimeouts;
         this.shutdownQuietPeriod = shutdownQuietPeriod;
         this.shutdownTimeout = shutdownTimeout;
 
@@ -585,6 +589,11 @@ public Duration getReadTimeout() {
         return readTimeout;
     }
 
+    @Override
+    public boolean isUseEventLoopTimeouts() {
+        return useEventLoopTimeouts;
+    }
+
     @Override
     public Duration getShutdownQuietPeriod() {
         return shutdownQuietPeriod;
@@ -958,6 +967,7 @@ public static class Builder {
         private Duration connectTimeout = defaultConnectTimeout();
         private Duration requestTimeout = defaultRequestTimeout();
         private Duration readTimeout = defaultReadTimeout();
+        private boolean useEventLoopTimeouts = defaultUseEventLoopTimeouts();
         private Duration shutdownQuietPeriod = defaultShutdownQuietPeriod();
         private Duration shutdownTimeout = defaultShutdownTimeout();
 
@@ -1064,6 +1074,7 @@ public Builder(AsyncHttpClientConfig config) {
             connectTimeout = config.getConnectTimeout();
             requestTimeout = config.getRequestTimeout();
             readTimeout = config.getReadTimeout();
+            useEventLoopTimeouts = config.isUseEventLoopTimeouts();
             shutdownQuietPeriod = config.getShutdownQuietPeriod();
             shutdownTimeout = config.getShutdownTimeout();
 
@@ -1355,6 +1366,17 @@ public Builder setReadTimeout(Duration readTimeout) {
             return this;
         }
 
+        /**
+         * @param useEventLoopTimeouts whether to arm request and read timeouts on an event loop instead of on
+         *                             the client's timer; see {@link AsyncHttpClientConfig#isUseEventLoopTimeouts()}
+         *                             for the trade-off this makes
+         * @return this
+         */
+        public Builder setUseEventLoopTimeouts(boolean useEventLoopTimeouts) {
+            this.useEventLoopTimeouts = useEventLoopTimeouts;
+            return this;
+        }
+
         public Builder setShutdownQuietPeriod(Duration shutdownQuietPeriod) {
             this.shutdownQuietPeriod = shutdownQuietPeriod;
             return this;
@@ -1764,6 +1786,7 @@ public DefaultAsyncHttpClientConfig build() {
                     connectTimeout,
                     requestTimeout,
                     readTimeout,
+                    useEventLoopTimeouts,
                     shutdownQuietPeriod,
                     shutdownTimeout,
                     keepAlive,

client/src/main/java/org/asynchttpclient/config/AsyncHttpClientConfigDefaults.java

@@ -41,6 +41,7 @@ public final class AsyncHttpClientConfigDefaults {
     public static final String CONNECTION_POOL_CLEANER_PERIOD_CONFIG = "connectionPoolCleanerPeriod";
     public static final String READ_TIMEOUT_CONFIG = "readTimeout";
     public static final String REQUEST_TIMEOUT_CONFIG = "requestTimeout";
+    public static final String USE_EVENT_LOOP_TIMEOUTS_CONFIG = "useEventLoopTimeouts";
     public static final String CONNECTION_TTL_CONFIG = "connectionTtl";
     public static final String FOLLOW_REDIRECT_CONFIG = "followRedirect";
     public static final String MAX_REDIRECTS_CONFIG = "maxRedirects";
@@ -154,6 +155,10 @@ public static Duration defaultRequestTimeout() {
         return AsyncHttpClientConfigHelper.getAsyncHttpClientConfig().getDuration(ASYNC_CLIENT_CONFIG_ROOT + REQUEST_TIMEOUT_CONFIG);
     }
 
+    public static boolean defaultUseEventLoopTimeouts() {
+        return AsyncHttpClientConfigHelper.getAsyncHttpClientConfig().getBoolean(ASYNC_CLIENT_CONFIG_ROOT + USE_EVENT_LOOP_TIMEOUTS_CONFIG);
+    }
+
     public static Duration defaultConnectionTtl() {
         return AsyncHttpClientConfigHelper.getAsyncHttpClientConfig().getDuration(ASYNC_CLIENT_CONFIG_ROOT + CONNECTION_TTL_CONFIG);
     }

client/src/main/java/org/asynchttpclient/netty/NettyResponseFuture.java

@@ -466,6 +466,11 @@ public void setTimeoutsHolder(TimeoutsHolder timeoutsHolder) {
         if (ref != null) {
             ref.cancel();
         }
+        if (timeoutsHolder != null) {
+            // Armed here rather than by the caller: a holder can run its timeout the moment it is armed, so it
+            // has to be reachable from this future first, and no caller can then install one that never arms.
+            timeoutsHolder.start();
+        }
     }
 
     public boolean isInAuth() {

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

@@ -124,6 +124,11 @@ public void onSuccess(Channel channel, InetSocketAddress remoteAddress) {
         // mid-handshake could not close the socket, stranding it until handshakeTimeout (issue #2189).
         future.attachChannel(channel, false);
 
+        // The timeouts were armed before there was a channel to arm them on; hand them the one the exchange
+        // ended up with. This listener runs on that channel's own loop, so the move needs no wakeup, and from
+        // here on an expiry runs on the thread that would have to close the socket.
+        timeoutsHolder.rehomeOn(channel.eventLoop());
+
         Request request = future.getTargetRequest();
         Uri uri = request.getUri();
         // don't set a null resolved address - if the remoteAddress is null we keep

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

@@ -82,6 +82,7 @@
 import org.asynchttpclient.resolver.RequestHostnameResolver;
 import org.asynchttpclient.uri.Uri;
 import org.asynchttpclient.ws.WebSocketUpgradeHandler;
+import org.jetbrains.annotations.Nullable;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -401,15 +402,17 @@ private <T> ListenableFuture<T> sendRequestWithOpenChannel(NettyResponseFuture<T
             return future;
         }
 
+        future.setChannelState(ChannelState.POOLED);
+        // Before the timeout is armed, not after: an expiry reaches the channel only through the future, and on
+        // an event loop a short enough deadline can be delivered before the next statement would have run.
+        future.attachChannel(channel, false);
+
         SocketAddress channelRemoteAddress = channel.remoteAddress();
         if (channelRemoteAddress != null) {
             // otherwise, bad luck, the channel was closed, see bellow
-            scheduleRequestTimeout(future, (InetSocketAddress) channelRemoteAddress);
+            scheduleRequestTimeout(future, (InetSocketAddress) channelRemoteAddress, channel);
         }
 
-        future.setChannelState(ChannelState.POOLED);
-        future.attachChannel(channel, false);
-
         if (LOGGER.isDebugEnabled()) {
             HttpRequest httpRequest = future.getNettyRequest().getHttpRequest();
             LOGGER.debug("Using open Channel {} for {} '{}'", channel, httpRequest.method(), httpRequest.uri());
@@ -1080,12 +1083,36 @@ private static void configureTransferAdapter(AsyncHandler<?> handler, HttpReques
 
     private void scheduleRequestTimeout(NettyResponseFuture<?> nettyResponseFuture,
                                         InetSocketAddress originalRemoteAddress) {
+        scheduleRequestTimeout(nettyResponseFuture, originalRemoteAddress, null);
+    }
+
+    /**
+     * @param channel the channel the exchange will run on when it is already known, so the timeout can be armed
+     *                on the loop that owns it. Null on the connect path: the timeout is armed before the channel
+     *                exists, deliberately, so that it also bounds address resolution and the connect itself, and
+     *                {@code TimeoutsHolder#rehomeOn} moves it onto the loop once there is one.
+     */
+    private void scheduleRequestTimeout(NettyResponseFuture<?> nettyResponseFuture,
+                                        InetSocketAddress originalRemoteAddress,
+                                        @Nullable Channel channel) {
         nettyResponseFuture.touch();
-        TimeoutsHolder timeoutsHolder = new TimeoutsHolder(nettyTimer, nettyResponseFuture, this, config,
-                originalRemoteAddress);
+        TimeoutsHolder timeoutsHolder = new TimeoutsHolder(nettyTimer, timeoutExecutor(channel), nettyResponseFuture,
+                this, config, originalRemoteAddress);
+        // Arms the timeout as a part of installing the holder, which is why the pooled path attaches the
+        // channel first: an expiry that lands immediately reaches the channel only through the future.
         nettyResponseFuture.setTimeoutsHolder(timeoutsHolder);
     }
 
+    /**
+     * The loop to arm an exchange's timeouts on, or null to leave them on the client's timer. Only ever the
+     * exchange's own channel's loop: any other loop would be woken by an entry it has no interest in, and the
+     * group's chooser hands out channels from the same counter, so drawing from it here would shift which loops
+     * connections land on.
+     */
+    private @Nullable EventExecutor timeoutExecutor(@Nullable Channel channel) {
+        return config.isUseEventLoopTimeouts() && channel != null ? channel.eventLoop() : null;
+    }
+
     private static void scheduleReadTimeout(NettyResponseFuture<?> nettyResponseFuture) {
         TimeoutsHolder timeoutsHolder = nettyResponseFuture.getTimeoutsHolder();
         if (timeoutsHolder != null) {

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

@@ -22,7 +22,7 @@
 
 import static org.asynchttpclient.util.DateUtils.unpreciseMillisTime;
 
-public class ReadTimeoutTimerTask extends TimeoutTimerTask {
+public class ReadTimeoutTimerTask extends TimeoutTimerTask implements Runnable {
 
     private final long readTimeout;
 
@@ -31,6 +31,15 @@ public class ReadTimeoutTimerTask extends TimeoutTimerTask {
         this.readTimeout = readTimeout;
     }
 
+    /**
+     * The event-loop entry point. Nothing below reads the {@link Timeout}, which is the timer's own handle on
+     * this task and something an event loop has no equivalent of, so both entry points share one body.
+     */
+    @Override
+    public void run() {
+        run(null);
+    }
+
     @Override
     public void run(Timeout timeout) {
         if (done.getAndSet(true) || requestSender.isClosed()) {

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

@@ -22,7 +22,7 @@
 
 import static org.asynchttpclient.util.DateUtils.unpreciseMillisTime;
 
-public class RequestTimeoutTimerTask extends TimeoutTimerTask {
+public class RequestTimeoutTimerTask extends TimeoutTimerTask implements Runnable {
 
     private final long requestTimeout;
 
@@ -34,6 +34,15 @@ public class RequestTimeoutTimerTask extends TimeoutTimerTask {
         this.requestTimeout = requestTimeout;
     }
 
+    /**
+     * The event-loop entry point. Nothing below reads the {@link Timeout}, which is the timer's own handle on
+     * this task and something an event loop has no equivalent of, so both entry points share one body.
+     */
+    @Override
+    public void run() {
+        run(null);
+    }
+
     @Override
     public void run(Timeout timeout) {
         if (done.getAndSet(true) || requestSender.isClosed()) {

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

@@ -15,16 +15,27 @@
  */
 package org.asynchttpclient.netty.timeout;
 
+import io.netty.util.Timeout;
 import io.netty.util.TimerTask;
+import io.netty.util.concurrent.ScheduledFuture;
 import org.asynchttpclient.netty.NettyResponseFuture;
 import org.asynchttpclient.netty.request.NettyRequestSender;
+import org.jetbrains.annotations.Nullable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.net.InetSocketAddress;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+/**
+ * A timeout that can be armed either on a {@link io.netty.util.Timer} or on an event loop; which one an
+ * exchange uses is {@link org.asynchttpclient.AsyncHttpClientConfig#isUseEventLoopTimeouts()}. An event loop
+ * schedules {@link Runnable}s, so each subclass implements that as a second entry point into the same body.
+ * This class stays a {@link TimerTask} alone: {@code Runnable#run} declares no checked exception, so a
+ * {@code run()} here would have to catch what {@link TimerTask#run(Timeout)} declares and no subclass throws.
+ */
 public abstract class TimeoutTimerTask implements TimerTask {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(TimeoutTimerTask.class);
@@ -33,13 +44,67 @@ public abstract class TimeoutTimerTask implements TimerTask {
     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;
 
     TimeoutTimerTask(NettyResponseFuture<?> nettyResponseFuture, NettyRequestSender requestSender, TimeoutsHolder timeoutsHolder) {
         this.nettyResponseFuture = nettyResponseFuture;
         this.requestSender = requestSender;
         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;
+    }
+
+    void armedOn(ScheduledFuture<?> handle) {
+        timerHandle = null;
+        loopHandle = handle;
+    }
+
+    /**
+     * Cancels the scheduled entry this task was armed on, if any. Never interrupts: on the event-loop path the
+     * task may be running on the very thread this is called from, and nothing in it answers interruption.
+     *
+     * @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();
+        }
+        ScheduledFuture<?> scheduled = loopHandle;
+        if (scheduled != null) {
+            loopHandle = null;
+            try {
+                return scheduled.cancel(false);
+            } catch (RejectedExecutionException e) {
+                // Cancelling from off the loop enqueues the removal, which a loop that is already shutting down
+                // rejects. The entry dies with the loop either way, and this runs under
+                // ListenableFuture#cancel, which has never thrown for a client that is closing.
+                LOGGER.debug("Event loop rejected a timeout cancellation", e);
+                return false;
+            }
+        }
+        return false;
+    }
+
+    /**
+     * Whether this task has been claimed, either by firing or by {@link #clean()}. Stands in for the
+     * scheduler's own already-expired flag, which the two schedulers spell differently, and is if anything the
+     * more precise of the two: it flips when {@code run} is entered rather than when the entry is marked.
+     */
+    boolean isClaimed() {
+        return done.get();
+    }
+
     void expire(String message, long time) {
         LOGGER.debug("{} for {} after {} ms", message, nettyResponseFuture, time);
         requestSender.abort(nettyResponseFuture.channel(), nettyResponseFuture, new TimeoutException(message));