@@ -326,31 +326,21 @@ class Scheduled extends State {
 
     @Override
     public void onStart() {
-      try {
-        stopHook = timer.onStop(this::onTimerStopping);
-        // Wraps go innermost so the captured gRPC + OpenTelemetry contexts are re-established at
-        // the moment the body runs on the op executor.
-        future =
-            timer.newTimeout(
-                () ->
-                    grpcContext
-                        .wrap(() -> otelContext.wrap(() -> onStateChange(new Idle())).run())
-                        .run(),
-                context.getExecutor(),
-                Durations.toMillis(retryDelay),
-                TimeUnit.MILLISECONDS);
-      } catch (IllegalStateException e) {
-        // Timer was stopped between Active.onClose deciding to retry and this task running on the
-        // op executor. Race window is narrow (post-drain shutdown), but cover it cleanly so the
-        // op-executor uncaught handler does not have to. onExit will release the stopHook.
-        onStateChange(
-            new Done(
-                VRpcResult.createRejectedError(
-                    Status.CANCELLED
-                        .withDescription(
-                            "Executor shutting down, can't schedule operation for retry.")
-                        .withCause(e))));
-      }
+      // If the timer is already stopped, onStop fires the hook synchronously here — which
+      // enqueues an onTimerStopping body on the op executor — and newTimeout returns a
+      // pre-cancelled Timeout. The queued onTimerStopping then drives us to Done(CANCELLED).
+      stopHook = timer.onStop(this::onTimerStopping);
+      // Wraps go innermost so the captured gRPC + OpenTelemetry contexts are re-established at
+      // the moment the body runs on the op executor.
+      future =
+          timer.newTimeout(
+              () ->
+                  grpcContext
+                      .wrap(() -> otelContext.wrap(() -> onStateChange(new Idle())).run())
+                      .run(),
+              context.getExecutor(),
+              Durations.toMillis(retryDelay),
+              TimeUnit.MILLISECONDS);
     }
 
     // Invoked from BigtableTimer.stop on the close thread. Trampoline back to the op executor so
 
@@ -23,6 +23,7 @@
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.Executor;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 import java.util.logging.Level;
 import java.util.logging.Logger;
@@ -50,14 +51,32 @@ public final class HashedWheelTimer implements BigtableTimer {
   // Bound transfers per tick so a sustained insert flood can't starve bucket processing.
   private static final int MAX_PENDING_TRANSFERS_PER_TICK = 100_000;
 
+  // Returned from newTimeout when the timer is already stopped. Pre-cancelled so callers that
+  // poll isCancelled() see "done" and skip cleanup paths; cancel() is a no-op.
+  private static final Timeout DEAD_TIMEOUT =
+      new Timeout() {
+        @Override
+        public boolean cancel() {
+          return false;
+        }
+
+        @Override
+        public boolean isCancelled() {
+          return true;
+        }
+      };
+
   private final HashedWheelBucket[] wheel = new HashedWheelBucket[TICKS_PER_WHEEL];
   private final Queue<HashedWheelTimeout> pendingTimeouts = new ConcurrentLinkedQueue<>();
   private final Set<Runnable> stopHooks = ConcurrentHashMap.newKeySet();
 
   private final Thread worker;
   private final long startNanos;
 
-  private volatile boolean stopped = false;
+  // Single-shot transition from false to true; CAS in stop() pairs with .get() in newTimeout/
+  // onStop so racing register-ers either land before stop()'s drain (and are caught by it) or
+  // observe true and self-resolve.
+  private final AtomicBoolean stopped = new AtomicBoolean(false);
 
   public HashedWheelTimer(String name) {
     for (int i = 0; i < TICKS_PER_WHEEL; i++) {
@@ -75,46 +94,73 @@ public HashedWheelTimer(String name) {
 
   @Override
   public Timeout newTimeout(Runnable task, Executor executor, long delay, TimeUnit unit) {
-    if (stopped) {
-      throw new IllegalStateException("timer stopped");
+    if (stopped.get()) {
+      return DEAD_TIMEOUT;
     }
     long delayNanos = Math.max(0L, unit.toNanos(delay));
     long deadlineNanos = (System.nanoTime() - startNanos) + delayNanos;
     HashedWheelTimeout timeout = new HashedWheelTimeout(task, executor, deadlineNanos);
     pendingTimeouts.add(timeout);
+    // If stop() raced after our get(), its drain loop will catch this entry and cancel it.
     return timeout;
   }
 
   @Override
   public Registration onStop(Runnable hook) {
-    if (stopped) {
-      throw new IllegalStateException("timer stopped");
+    if (stopped.get()) {
+      hook.run();
+      return NO_OP_REGISTRATION;
     }
     stopHooks.add(hook);
+    if (stopped.get()) {
+      // Raced with stop(). Either its drain already consumed and invoked our hook (set.remove
+      // returns false → no-op), or our add landed after its drain (set.remove returns true →
+      // we own it and must fire it ourselves). Either way the caller sees the hook run.
+      if (stopHooks.remove(hook)) {
+        hook.run();
+      }
+      return NO_OP_REGISTRATION;
+    }
     return () -> stopHooks.remove(hook);
   }
 
+  private static final Registration NO_OP_REGISTRATION = () -> {};
+
   @Override
   public void stop() {
-    if (stopped) {
+    if (!stopped.compareAndSet(false, true)) {
       return;
     }
-    stopped = true;
-    Set<Runnable> hooks = new HashSet<>(stopHooks);
-    stopHooks.clear();
-    for (Runnable hook : hooks) {
-      try {
-        hook.run();
-      } catch (Throwable t) {
-        LOG.log(Level.WARNING, "stop hook threw; continuing", t);
+    // Drain both queues in a loop. In-flight register-ers that read stopped==false before our
+    // CAS will eventually finish their add and land in stopHooks/pendingTimeouts; the loop
+    // keeps draining until no new entries appear. Readers that arrive after our CAS observe
+    // stopped==true and skip the add entirely.
+    while (true) {
+      Set<Runnable> hooks = new HashSet<>(stopHooks);
+      stopHooks.removeAll(hooks);
+      for (Runnable hook : hooks) {
+        try {
+          hook.run();
+        } catch (Throwable t) {
+          LOG.log(Level.WARNING, "stop hook threw; continuing", t);
+        }
+      }
+      int drained = 0;
+      HashedWheelTimeout t;
+      while ((t = pendingTimeouts.poll()) != null) {
+        t.cancel();
+        drained++;
+      }
+      if (hooks.isEmpty() && drained == 0) {
+        break;
       }
     }
     worker.interrupt();
   }
 
   private void workerLoop() {
     long tick = 0;
-    while (!stopped) {
+    while (!stopped.get()) {
       if (waitForNextTick(tick) < 0) {
         return;
       }
@@ -133,14 +179,14 @@ private long waitForNextTick(long tick) {
       if (sleepNanos <= 0) {
         return elapsed;
       }
-      if (stopped) {
+      if (stopped.get()) {
         return -1;
       }
       long sleepMs = (sleepNanos + 999_999L) / 1_000_000L;
       try {
         Thread.sleep(sleepMs);
       } catch (InterruptedException e) {
-        if (stopped) {
+        if (stopped.get()) {
           return -1;
         }
         // Spurious interrupt; recompute and keep sleeping.
 
@@ -16,7 +16,6 @@
 package com.google.cloud.bigtable.data.v2.internal.session;
 
 import static com.google.common.truth.Truth.assertThat;
-import static org.junit.jupiter.api.Assertions.assertThrows;
 
 import com.google.cloud.bigtable.data.v2.internal.session.BigtableTimer.Registration;
 import com.google.common.util.concurrent.MoreExecutors;
@@ -68,7 +67,7 @@ void cancel_before_fire_prevents_execution() throws Exception {
   }
 
   @Test
-  void stop_runs_all_hooks_on_caller_thread_and_rejects_inserts() {
+  void stop_runs_all_hooks_on_caller_thread_then_newTimeout_returns_dead() {
     timer = new HashedWheelTimer("wheel-test");
     Thread caller = Thread.currentThread();
     List<Thread> ran = new CopyOnWriteArrayList<>();
@@ -82,9 +81,39 @@ void stop_runs_all_hooks_on_caller_thread_and_rejects_inserts() {
     for (Thread t : ran) {
       assertThat(t).isSameInstanceAs(caller);
     }
-    assertThrows(
-        IllegalStateException.class,
-        () -> timer.newTimeout(() -> {}, DIRECT, 1, TimeUnit.MILLISECONDS));
+    // After stop, newTimeout returns a pre-cancelled Timeout that never fires.
+    BigtableTimer.Timeout t = timer.newTimeout(() -> {}, DIRECT, 1, TimeUnit.MILLISECONDS);
+    assertThat(t.isCancelled()).isTrue();
+    timer = null;
+  }
+
+  @Test
+  void onStop_after_stop_fires_hook_synchronously() {
+    timer = new HashedWheelTimer("wheel-test");
+    timer.stop();
+    Thread caller = Thread.currentThread();
+    List<Thread> ran = new CopyOnWriteArrayList<>();
+
+    Registration reg = timer.onStop(() -> ran.add(Thread.currentThread()));
+
+    assertThat(ran).hasSize(1);
+    assertThat(ran.get(0)).isSameInstanceAs(caller);
+    // Registration is still safe to call; should be a no-op.
+    reg.unregister();
+    timer = null;
+  }
+
+  @Test
+  void double_stop_does_not_re_run_hooks() {
+    timer = new HashedWheelTimer("wheel-test");
+    AtomicInteger fires = new AtomicInteger();
+
+    timer.onStop(fires::incrementAndGet);
+
+    timer.stop();
+    timer.stop();
+
+    assertThat(fires.get()).isEqualTo(1);
     timer = null;
   }