fix(runtime-broker): keep the HTTP exchange cancel off the deadline completion path (#13562)

When a transport call hits its deadline, the cancel of the underlying HTTP
exchange ran synchronously on the shared CompletableFuture delay thread and,
because of how nested completions propagate, before the caller's own
continuation. The caller therefore paid for the whole HttpClient cancel and
socket close, which on Windows occasionally takes about 3 s and made
ManagedCsiAcknowledgementHttpTransportTest.totalDeadlineIncludesAResponseBodyThatNeverFinishes
exceed its 3 s bound (runs 37543910341, 37542230965, 37558689200). A slow
cancel on that thread also delays every other orTimeout in the JVM.

Run the cancel on an async continuation instead, at both deadline sites, and
add a regression test that injects a blocking cancel and asserts the caller is
released first. Deadline semantics, error codes and the connection close are
unchanged.

packages/sdk-java/runtime-broker/src/main/java/com/alibaba/qwen/code/runtimebroker/HttpRuntimeTransport.java

@@ -219,7 +219,10 @@ public final class HttpRuntimeTransport implements RuntimeTransport {
                     }
                     throw unavailable(cause);
                 });
-        returned.whenComplete((value, error) -> {
+        // A deadline completes the stage on the shared CompletableFuture delay
+        // thread, and cancelling the exchange closes a socket: keep that off
+        // the caller's completion path and off the timer thread.
+        returned.whenCompleteAsync((value, error) -> {
             if (error != null || returned.isCancelled()) {
                 exchange.cancel(true);
                 result.cancel(false);
@@ -1109,7 +1112,10 @@ public final class HttpRuntimeTransport implements RuntimeTransport {
                     }
                     throw unavailable(cause);
                 });
-        returned.whenComplete((value, error) -> {
+        // A deadline completes the stage on the shared CompletableFuture delay
+        // thread, and cancelling the exchange closes a socket: keep that off
+        // the caller's completion path and off the timer thread.
+        returned.whenCompleteAsync((value, error) -> {
             if (error != null || returned.isCancelled()) {
                 exchange.cancel(true);
                 result.cancel(false);

packages/sdk-java/runtime-broker/src/test/java/com/alibaba/qwen/code/runtimebroker/ManagedCsiAcknowledgementHttpTransportTest.java

@@ -11,19 +11,29 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
 import com.sun.net.httpserver.HttpServer;
 import java.net.InetSocketAddress;
 import java.net.URI;
+import java.net.Authenticator;
+import java.net.CookieHandler;
+import java.net.ProxySelector;
 import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
 import java.math.BigDecimal;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executor;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLParameters;
 import org.junit.jupiter.api.Test;
 
 class ManagedCsiAcknowledgementHttpTransportTest {
@@ -198,6 +208,34 @@ class ManagedCsiAcknowledgementHttpTransportTest {
         }
     }
 
+    @Test
+    void deadlineFailureDoesNotWaitForTheExchangeCancel() throws Exception {
+        var fixture = fixtures();
+        var callerReturned = new CountDownLatch(1);
+        var cancelFinished = new CountDownLatch(1);
+        var cancelSawCallerReturn = new AtomicBoolean();
+        var client = new StalledExchangeClient(() -> {
+            try {
+                cancelSawCallerReturn.set(callerReturned.await(5, TimeUnit.SECONDS));
+            } catch (InterruptedException interrupted) {
+                Thread.currentThread().interrupt();
+            } finally {
+                cancelFinished.countDown();
+            }
+        });
+        var boot = map(fixture.get("boot"));
+        var transport = new HttpRuntimeTransport(client, Duration.ofMillis(200));
+        // The endpoint is never contacted: the stalled client answers nothing until the deadline fires.
+        var stage = transport.acknowledgeCsi(
+                ManagedCsiAcknowledgementProtocolTest.lease(boot, URI.create("http://127.0.0.1:9")),
+                ManagedCsiAcknowledgementProtocolTest.session(fixture), boot, map(fixture.get("expectedPod")),
+                map(fixture.get("request")), map(map(fixture.get("response")).get("captureIdentity")));
+        assertEquals("managed_runtime_unavailable", failure(() -> stage).getCode());
+        callerReturned.countDown();
+        assertTrue(cancelFinished.await(5, TimeUnit.SECONDS));
+        assertTrue(cancelSawCallerReturn.get(), "the deadline failure waited for the exchange cancel");
+    }
+
     private static RuntimeBrokerException failure(java.util.function.Supplier<CompletionStage<Map<String, Object>>> call) {
         return assertInstanceOf(RuntimeBrokerException.class,
                 assertThrows(CompletionException.class, () -> call.get().toCompletableFuture().join()).getCause());
@@ -206,6 +244,83 @@ class ManagedCsiAcknowledgementHttpTransportTest {
     private record Reply(int status, byte[] body, boolean headers, boolean encoded) {
     }
 
+    /** Never answers a request; runs {@code onCancel} inside the exchange cancel, on the cancelling thread. */
+    private static final class StalledExchangeClient extends HttpClient {
+        private final Runnable onCancel;
+
+        StalledExchangeClient(Runnable onCancel) {
+            this.onCancel = onCancel;
+        }
+
+        @Override
+        public <T> CompletableFuture<HttpResponse<T>> sendAsync(HttpRequest request,
+                HttpResponse.BodyHandler<T> handler) {
+            return new CompletableFuture<>() {
+                @Override
+                public boolean cancel(boolean mayInterruptIfRunning) {
+                    onCancel.run();
+                    return super.cancel(mayInterruptIfRunning);
+                }
+            };
+        }
+
+        @Override
+        public <T> CompletableFuture<HttpResponse<T>> sendAsync(HttpRequest request,
+                HttpResponse.BodyHandler<T> handler, HttpResponse.PushPromiseHandler<T> pushPromiseHandler) {
+            return sendAsync(request, handler);
+        }
+
+        @Override
+        public <T> HttpResponse<T> send(HttpRequest request, HttpResponse.BodyHandler<T> handler) {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public Redirect followRedirects() {
+            return Redirect.NEVER;
+        }
+
+        @Override
+        public Version version() {
+            return Version.HTTP_1_1;
+        }
+
+        @Override
+        public Optional<CookieHandler> cookieHandler() {
+            return Optional.empty();
+        }
+
+        @Override
+        public Optional<Duration> connectTimeout() {
+            return Optional.empty();
+        }
+
+        @Override
+        public Optional<ProxySelector> proxy() {
+            return Optional.empty();
+        }
+
+        @Override
+        public SSLContext sslContext() {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public SSLParameters sslParameters() {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public Optional<Authenticator> authenticator() {
+            return Optional.empty();
+        }
+
+        @Override
+        public Optional<Executor> executor() {
+            return Optional.empty();
+        }
+    }
+
     private static final class UnsupportedTransport implements RuntimeTransport {
         @Override
         public CompletionStage<Void> acquire(RuntimeLease lease, RuntimeSession session) {