From 3c6029baea1cc7a402a9da83a3dc9f9d8cdbd226 Mon Sep 17 00:00:00 2001 From: Hanabi <317387557+Hanabi9248@users.noreply.github.com> Date: Thu, 29 Oct 2026 01:17:48 +0800 Subject: [PATCH] fix: complete waiting HTTP requests after connection failures Release the existing semaphore after HttpUtils.async() prepares its failure response, matching the successful-response path. Otherwise a completed OkHttp failure leaves the public caller blocked indefinitely. Cover refused loopback connections, disconnects before HTTP headers, and successful async/sync requests using the complete production class and real OkHttp networking. Assisted-by: Codex 0.165.0-alpha.1.2 Signed-off-by: Hanabi <317387557+Hanabi9248@users.noreply.github.com> --- .../java/com/highgo/cloud/util/HttpUtils.java | 1 + .../HttpAsyncFailureRegression.java | 179 ++++++++++++++++++ 2 files changed, 180 insertions(+) create mode 100644 tests/regression/HttpAsyncFailureRegression.java diff --git a/common/src/main/java/com/highgo/cloud/util/HttpUtils.java b/common/src/main/java/com/highgo/cloud/util/HttpUtils.java index e78aa24..7cfa281 100644 --- a/common/src/main/java/com/highgo/cloud/util/HttpUtils.java +++ b/common/src/main/java/com/highgo/cloud/util/HttpUtils.java @@ -217,6 +217,7 @@ public String async() { @Override public void onFailure(Call call, IOException e) { buffer.append("请求出错：").append(e.getMessage()); + getSemaphoreInstance().release(); } @Override diff --git a/tests/regression/HttpAsyncFailureRegression.java b/tests/regression/HttpAsyncFailureRegression.java new file mode 100644 index 0000000..35e38ba --- /dev/null +++ b/tests/regression/HttpAsyncFailureRegression.java @@ -0,0 +1,179 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ + +import com.highgo.cloud.util.HttpUtils; +import com.sun.net.httpserver.HttpExchange; +import com.sun.net.httpserver.HttpServer; +import okhttp3.OkHttpClient; + +import java.io.OutputStream; +import java.lang.reflect.Field; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.ServerSocket; +import java.nio.charset.StandardCharsets; +import java.util.concurrent.Callable; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicReference; + +/** + * Calls the complete production HTTP builder and real OkHttp against loopback sockets. + * No client/call/callback/semaphore or logger is replaced. + */ +public final class HttpAsyncFailureRegression { + private static int passed; + private static int failed; + + private HttpAsyncFailureRegression() { + } + + public static void main(String[] args) throws Exception { + HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + ExecutorService serverWorker = Executors.newSingleThreadExecutor(task -> { + Thread thread = new Thread(task, "http-regression-server"); + thread.setDaemon(true); + return thread; + }); + server.setExecutor(serverWorker); + server.createContext("/disconnect", HttpExchange::close); + server.createContext("/success", exchange -> { + byte[] bytes = "actual-http-response".getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().set("Content-Type", "text/plain; charset=UTF-8"); + exchange.sendResponseHeaders(200, bytes.length); + try (OutputStream output = exchange.getResponseBody()) { + output.write(bytes); + } finally { + exchange.close(); + } + }); + server.start(); + String base = "http://127.0.0.1:" + server.getAddress().getPort(); + try { + check("refused connection returns the prepared failure text", () -> { + int unusedPort; + try (ServerSocket reservation = new ServerSocket( + 0, 1, InetAddress.getByName("127.0.0.1"))) { + unusedPort = reservation.getLocalPort(); + } + String response = bounded(() -> HttpUtils.builder() + .url("http://127.0.0.1:" + unusedPort + "/unavailable").get().async()); + requireFailureText(response); + }); + check("disconnect before HTTP headers returns failure text", () -> { + String response = bounded(() -> HttpUtils.builder() + .url(base + "/disconnect").get().async()); + requireFailureText(response); + }); + check("successful asynchronous response still returns its body", () -> { + String response = bounded(() -> HttpUtils.builder() + .url(base + "/success").get().async()); + require("actual-http-response".equals(response), "unexpected async body: " + response); + }); + check("synchronous response behavior is unchanged", () -> { + String response = bounded(() -> HttpUtils.builder() + .url(base + "/success").get().sync()); + require("actual-http-response".equals(response), "unexpected sync body: " + response); + }); + } finally { + server.stop(0); + serverWorker.shutdownNow(); + closeActualClient(); + } + System.out.println("RESULT passed=" + passed + " failed=" + failed); + if (failed != 0) { + throw new AssertionError("HTTP completion regressions failed: " + failed); + } + } + + private static void requireFailureText(String response) { + require(response.startsWith("请求出错："), "missing existing failure prefix: " + response); + require(response.length() > "请求出错：".length(), "missing actual IOException detail"); + } + + private static String bounded(Callable operation) throws Exception { + ExecutorService caller = Executors.newSingleThreadExecutor(task -> { + Thread thread = new Thread(task, "http-regression-caller"); + thread.setDaemon(true); + return thread; + }); + AtomicReference interruptedResult = new AtomicReference<>(); + Future result = caller.submit(() -> { + String value = operation.call(); + interruptedResult.set(value); + return value; + }); + try { + return result.get(8, TimeUnit.SECONDS); + } catch (TimeoutException timeout) { + result.cancel(true); + caller.shutdownNow(); + if (!caller.awaitTermination(2, TimeUnit.SECONDS)) { + throw new AssertionError("HTTP caller did not terminate after timeout", timeout); + } + String prepared = interruptedResult.get(); + if (prepared == null || !prepared.startsWith("请求出错：")) { + throw new AssertionError("timeout without observed failure text; transport inconclusive", timeout); + } + throw new AssertionError("operation remained blocked after its real failure text: " + prepared, timeout); + } finally { + caller.shutdownNow(); + if (!caller.awaitTermination(2, TimeUnit.SECONDS)) { + throw new AssertionError("HTTP caller did not terminate after interruption"); + } + } + } + + private static void closeActualClient() throws Exception { + // Cleanup only: retrieve the production client; never replace it or its transport. + Field field = HttpUtils.class.getDeclaredField("okHttpClient"); + field.setAccessible(true); + OkHttpClient client = (OkHttpClient) field.get(null); + if (client != null) { + client.dispatcher().executorService().shutdownNow(); + client.connectionPool().evictAll(); + if (client.cache() != null) { + client.cache().close(); + } + } + } + + private static void require(boolean condition, String message) { + if (!condition) { + throw new AssertionError(message); + } + } + + private static void check(String name, CheckedRunnable operation) { + try { + operation.run(); + passed++; + System.out.println("PASS " + name); + } catch (Throwable error) { + failed++; + System.out.println("FAIL " + name + ": " + error); + error.printStackTrace(System.out); + } + } + + private interface CheckedRunnable { + void run() throws Exception; + } +}