fix: contain throttled request setup failures

core/src/main/java/com/datastax/dse/driver/internal/core/cql/continuous/ContinuousRequestHandlerBase.java

@@ -149,6 +149,8 @@ public abstract class ContinuousRequestHandlerBase<StatementT extends Request, R
   /** The callback selected to stream results back to the client. */
   private final CompletableFuture<NodeResponseCallback> chosenCallback = new CompletableFuture<>();
 
+  private final AtomicBoolean terminal = new AtomicBoolean();
+
   /**
    * How many speculative executions are currently running (including the initial execution). We
    * track this in order to know when to fail the request if all executions have reached the end of
@@ -283,6 +285,8 @@ public void onThrottleFailure(@NonNull RequestThrottlingException error) {
   }
 
   private boolean abortGlobalRequestOrChosenCallback(@NonNull Throwable error) {
+    terminal.set(true);
+    cancelGlobalTimeout();
     boolean completedChosenCallback = chosenCallback.completeExceptionally(error);
     if (!completedChosenCallback) {
       chosenCallback.thenAccept(callback -> callback.abort(error, false));
@@ -291,7 +295,16 @@ private boolean abortGlobalRequestOrChosenCallback(@NonNull Throwable error) {
   }
 
   public CompletionStage<ResultSetT> handle() {
-    globalTimeout = scheduleGlobalTimeout();
+    // Immediate admission happens in the continuous graph handler's constructor. If setup failed
+    // there, chosenCallback is already terminal and there is no live request to time out.
+    if (!terminal.get()) {
+      globalTimeout = scheduleGlobalTimeout();
+      // Admission can race with handle() after the check above but before globalTimeout is
+      // assigned. Ensure a synchronous terminal setup failure cannot leave that timeout behind.
+      if (terminal.get()) {
+        cancelGlobalTimeout();
+      }
+    }
     return fetchNextPage();
   }
 
@@ -370,7 +383,7 @@ private void sendRequest(
       }
     } else if (!chosenCallback.isDone()) {
       boolean writeSubmitted = false;
-      Throwable terminalPreWriteFailure = null;
+      Throwable terminalSetupFailure = null;
       NodeResponseCallback nodeResponseCallback = null;
       try {
         nodeResponseCallback =
@@ -392,26 +405,35 @@ private void sendRequest(
         writeSubmitted = true;
         writeFuture.addListener(nodeResponseCallback);
       } catch (Throwable t) {
-        if (!writeSubmitted && activeExecutionsCount.decrementAndGet() == 0) {
-          if (abortGlobalRequestOrChosenCallback(t)) {
-            terminalPreWriteFailure = t;
+        recordError(node, t);
+        if (activeExecutionsCount.decrementAndGet() == 0) {
+          if (abortGlobalRequestOrChosenCallback(t) && !(t instanceof CancellationException)) {
+            terminalSetupFailure = t;
           }
+        } else {
+          Loggers.warnWithException(
+              LOG, "[{}] Request setup failed, another execution is still active", logPrefix, t);
         }
-        throw t;
       } finally {
         if (!writeSubmitted) {
           if (nodeResponseCallback != null) {
             inFlightCallbacks.remove(nodeResponseCallback);
           }
           try {
             channel.cancelPreAcquireId();
-          } finally {
-            if (terminalPreWriteFailure != null) {
-              throttler.signalError(this, terminalPreWriteFailure);
+          } catch (Throwable cleanupFailure) {
+            if (terminalSetupFailure != null && terminalSetupFailure != cleanupFailure) {
+              terminalSetupFailure.addSuppressed(cleanupFailure);
+            } else {
+              Loggers.warnWithException(
+                  LOG, "[{}] Failed to cancel stream ID reservation", logPrefix, cleanupFailure);
             }
           }
         }
       }
+      if (terminalSetupFailure != null) {
+        throttler.signalError(this, terminalSetupFailure);
+      }
     } else {
       channel.cancelPreAcquireId();
     }

core/src/main/java/com/datastax/oss/driver/internal/core/adminrequest/ThrottledAdminRequestHandler.java

@@ -105,6 +105,7 @@ public static ThrottledAdminRequestHandler<ByteBuffer> prepare(
   private final long startTimeNanos;
   private final RequestThrottler throttler;
   private final SessionMetricUpdater metricUpdater;
+  private volatile boolean admitted;
   private final AtomicBoolean holdsExternalReservation;
 
   protected ThrottledAdminRequestHandler(
@@ -140,16 +141,20 @@ public CompletionStage<ResultT> start() {
       throttler.register(this);
     } catch (Throwable t) {
       cancelExternalReservation();
-      // Registration can fail before the throttler admits this request, so complete the result
-      // without calling this class's override, which would signal a permit that was never acquired.
-      super.setFinalError(t);
+      if (admitted) {
+        setFinalError(t);
+      } else {
+        // Registration failed before admission, so there is no throttler permit to release.
+        super.setFinalError(t);
+      }
       throw t;
     }
     return result;
   }
 
   @Override
   public void onThrottleReady(boolean wasDelayed) {
+    admitted = true;
     try {
       if (wasDelayed) {
         metricUpdater.updateTimer(
@@ -163,7 +168,6 @@ public void onThrottleReady(boolean wasDelayed) {
     } catch (Throwable t) {
       cancelExternalReservation();
       setFinalError(t);
-      throw t;
     }
   }
 
@@ -175,8 +179,6 @@ public void onThrottleFailure(@NonNull RequestThrottlingException error) {
   }
 
   private void cancelExternalReservation() {
-    // register() can invoke onThrottleReady() synchronously. If that callback throws, both
-    // onThrottleReady() and start() catch the same failure, so cancellation must be idempotent.
     if (holdsExternalReservation.compareAndSet(true, false)) {
       cancelCallerOwnedPreAcquireId();
     }
@@ -197,7 +199,7 @@ protected boolean setFinalError(Throwable error) {
     if (wasSet) {
       if (error instanceof DriverTimeoutException) {
         throttler.signalTimeout(this);
-      } else if (!(error instanceof RequestThrottlingException)) {
+      } else if (admitted) {
         throttler.signalError(this, error);
       }
     }