From 70ebcf98da95ef8c017a138991612d9c5cad62d8 Mon Sep 17 00:00:00 2001 From: Pavel Ptashyts <49400901+pavel-ptashyts@users.noreply.github.com> Date: Mon, 20 Jul 2026 23:24:25 +0200 Subject: [PATCH] Release throttle permits only once A completion callback can reentrantly trigger onThrowable when a reactive subscriber disposes and cancels its future. ReleasePermitOnComplete then released the same acquired permit from both terminal callbacks, inflating the semaphore and overflowing when its count reached Integer.MAX_VALUE. Guard the release atomically while continuing to forward both callbacks to the wrapped handler. Add a regression test reproducing the reported completion-to-cancellation callback sequence. Fixes #1797 Codex on behalf of Pavel Ptashyts Co-Authored-By: Codex --- .../filter/ReleasePermitOnComplete.java | 14 +++-- .../filter/ReleasePermitOnCompleteTest.java | 62 +++++++++++++++++++ 2 files changed, 72 insertions(+), 4 deletions(-) create mode 100644 client/src/test/java/org/asynchttpclient/filter/ReleasePermitOnCompleteTest.java diff --git a/client/src/main/java/org/asynchttpclient/filter/ReleasePermitOnComplete.java b/client/src/main/java/org/asynchttpclient/filter/ReleasePermitOnComplete.java index baf8585078..e1eba529f9 100644 --- a/client/src/main/java/org/asynchttpclient/filter/ReleasePermitOnComplete.java +++ b/client/src/main/java/org/asynchttpclient/filter/ReleasePermitOnComplete.java @@ -22,9 +22,11 @@ import java.util.HashSet; import java.util.Set; import java.util.concurrent.Semaphore; +import java.util.concurrent.atomic.AtomicBoolean; /** - * Wrapper for {@link AsyncHandler}s to release a permit on {@link AsyncHandler#onCompleted()}. This is done via a dynamic proxy to preserve all interfaces of the wrapped handler. + * Wrapper for {@link AsyncHandler}s to release a permit once on {@link AsyncHandler#onCompleted()} or + * {@link AsyncHandler#onThrowable(Throwable)}. This is done via a dynamic proxy to preserve all interfaces of the wrapped handler. */ public final class ReleasePermitOnComplete { @@ -33,10 +35,10 @@ private ReleasePermitOnComplete() { } /** - * Wrap handler to release the permit of the semaphore on {@link AsyncHandler#onCompleted()}. + * Wrap handler to release the semaphore permit once when the wrapped handler terminates. * * @param handler the handler to be wrapped - * @param available the Semaphore to be released when the wrapped handler is completed + * @param available the Semaphore to be released when the wrapped handler terminates * @param the handler result type * @return the wrapped handler */ @@ -45,6 +47,7 @@ public static AsyncHandler wrap(final AsyncHandler handler, final Sema Class handlerClass = handler.getClass(); ClassLoader classLoader = handlerClass.getClassLoader(); Class[] interfaces = allInterfaces(handlerClass); + AtomicBoolean permitReleased = new AtomicBoolean(); return (AsyncHandler) Proxy.newProxyInstance(classLoader, interfaces, (proxy, method, args) -> { try { @@ -53,7 +56,10 @@ public static AsyncHandler wrap(final AsyncHandler handler, final Sema switch (method.getName()) { case "onCompleted": case "onThrowable": - available.release(); + if (permitReleased.compareAndSet(false, true)) { + available.release(); + } + break; default: } } diff --git a/client/src/test/java/org/asynchttpclient/filter/ReleasePermitOnCompleteTest.java b/client/src/test/java/org/asynchttpclient/filter/ReleasePermitOnCompleteTest.java new file mode 100644 index 0000000000..b5c60d08e0 --- /dev/null +++ b/client/src/test/java/org/asynchttpclient/filter/ReleasePermitOnCompleteTest.java @@ -0,0 +1,62 @@ +/* + * Copyright (c) 2026 AsyncHttpClient Project. All rights reserved. + * + * Licensed 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. + */ +package org.asynchttpclient.filter; + +import org.asynchttpclient.AsyncCompletionHandler; +import org.asynchttpclient.AsyncHandler; +import org.asynchttpclient.Response; +import org.jetbrains.annotations.Nullable; +import org.junit.jupiter.api.Test; + +import java.util.concurrent.Semaphore; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class ReleasePermitOnCompleteTest { + + @Test + public void releasesPermitOnceWhenCompletionTriggersThrowable() { + Semaphore available = new Semaphore(Integer.MAX_VALUE); + assertTrue(available.tryAcquire()); + + AtomicReference> wrapped = new AtomicReference<>(); + AtomicInteger completedCalls = new AtomicInteger(); + AtomicInteger throwableCalls = new AtomicInteger(); + AsyncHandler handler = new AsyncCompletionHandler() { + @Override + public @Nullable Object onCompleted(@Nullable Response response) { + completedCalls.incrementAndGet(); + wrapped.get().onThrowable(new RuntimeException("cancelled during completion")); + return null; + } + + @Override + public void onThrowable(Throwable t) { + throwableCalls.incrementAndGet(); + } + }; + wrapped.set(ReleasePermitOnComplete.wrap(handler, available)); + + assertDoesNotThrow(() -> wrapped.get().onCompleted()); + assertEquals(Integer.MAX_VALUE, available.availablePermits()); + assertEquals(1, completedCalls.get()); + assertEquals(1, throwableCalls.get()); + } +}