change from java semaphore to custom AsynSemaphore

Author: MarketDataDev03

src/main/java/com/marketdata/sdk/internal/http/AsyncSemaphore.java

@@ -0,0 +1,95 @@
+package com.marketdata.sdk.internal.http;
+
+import java.util.ArrayDeque;
+import java.util.Deque;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Async-safe concurrency limiter. Replaces {@link java.util.concurrent.Semaphore} in the HTTP path
+ * so that {@code executeAsync} never parks the caller's thread when the pool is at capacity — it
+ * returns a {@link CompletableFuture} that completes when a permit is released by an in-flight
+ * request. See ADR-007 for the rationale.
+ *
+ * <p>Two invariants:
+ *
+ * <ol>
+ *   <li>Every permit is accounted for exactly once — it is either in {@link #availablePermits()}
+ *       (free), held by an in-flight caller (and will be released via {@link #release()}), or
+ *       pending in the waiter queue (and will be released by completing the waiter's future).
+ *   <li>{@link CompletableFuture#complete} of a transferred permit always runs <em>outside</em> the
+ *       lock. Completing a future runs the caller's attached callbacks synchronously on the
+ *       releasing thread, and we never want those running while our lock is held.
+ * </ol>
+ *
+ * <p>Cancelled or otherwise-completed waiters are skipped on {@link #release()} so a cancelled
+ * {@code acquire} doesn't burn a permit.
+ */
+public final class AsyncSemaphore {
+
+  private final Object lock = new Object();
+  private final Deque<CompletableFuture<Void>> waiters = new ArrayDeque<>();
+  private int available;
+
+  public AsyncSemaphore(int permits) {
+    if (permits < 0) {
+      throw new IllegalArgumentException("permits must be >= 0, was " + permits);
+    }
+    this.available = permits;
+  }
+
+  /**
+   * Asynchronously claim a permit.
+   *
+   * <p>Fast path: a permit is available, returns an already-completed future. Slow path: pool is
+   * exhausted, returns a pending future enqueued FIFO; it completes when some in-flight caller
+   * calls {@link #release()}. Either way, the caller's thread is never parked.
+   */
+  public CompletableFuture<Void> acquire() {
+    synchronized (lock) {
+      if (available > 0) {
+        available--;
+        return CompletableFuture.completedFuture(null);
+      }
+      CompletableFuture<Void> waiter = new CompletableFuture<>();
+      waiters.addLast(waiter);
+      return waiter;
+    }
+  }
+
+  /**
+   * Release a permit. If a live waiter is enqueued, the permit is transferred to it (its future is
+   * completed) without going through the counter. Otherwise the counter is incremented.
+   */
+  public void release() {
+    CompletableFuture<Void> next = null;
+    synchronized (lock) {
+      while (!waiters.isEmpty()) {
+        CompletableFuture<Void> w = waiters.pollFirst();
+        if (!w.isDone()) {
+          next = w;
+          break;
+        }
+      }
+      if (next == null) {
+        available++;
+      }
+    }
+    if (next != null) {
+      next.complete(null);
+    }
+  }
+
+  /** Permits not currently held nor pending in the queue. */
+  public int availablePermits() {
+    synchronized (lock) {
+      return available;
+    }
+  }
+
+  /** Number of pending waiters on the slow path. Useful for diagnostics and tests. */
+  public int queueLength() {
+    synchronized (lock) {
+      return waiters.size();
+    }
+  }
+}

src/main/java/com/marketdata/sdk/internal/http/HttpTransport.java

@@ -18,7 +18,6 @@
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
-import java.util.concurrent.Semaphore;
 import java.util.concurrent.atomic.AtomicReference;
 import org.jspecify.annotations.Nullable;
 
@@ -49,7 +48,7 @@ public final class HttpTransport implements AutoCloseable {
 
   private final HttpClient httpClient;
   private final ObjectMapper jsonMapper;
-  private final Semaphore concurrencyPermits;
+  private final AsyncSemaphore concurrencyPermits;
   private final AtomicReference<@Nullable RateLimits> latestRateLimits = new AtomicReference<>();
 
   private final String baseUrl;
@@ -59,18 +58,32 @@ public final class HttpTransport implements AutoCloseable {
 
   public HttpTransport(
       String baseUrl, String apiVersion, String userAgent, @Nullable String token) {
+    this(baseUrl, apiVersion, userAgent, token, defaultHttpClient());
+  }
+
+  // Package-private constructor used by tests to inject a stubbed HttpClient
+  // (e.g. one whose sendAsync throws synchronously, to verify permit release).
+  HttpTransport(
+      String baseUrl,
+      String apiVersion,
+      String userAgent,
+      @Nullable String token,
+      HttpClient httpClient) {
     this.baseUrl = baseUrl;
     this.apiVersion = apiVersion;
     this.userAgent = userAgent;
     this.token = token;
-    this.concurrencyPermits = new Semaphore(CONCURRENCY_LIMIT);
+    this.concurrencyPermits = new AsyncSemaphore(CONCURRENCY_LIMIT);
     this.jsonMapper = new ObjectMapper();
-    this.httpClient =
-        HttpClient.newBuilder()
-            .connectTimeout(CONNECT_TIMEOUT)
-            .version(HttpClient.Version.HTTP_2)
-            .followRedirects(HttpClient.Redirect.NORMAL)
-            .build();
+    this.httpClient = httpClient;
+  }
+
+  private static HttpClient defaultHttpClient() {
+    return HttpClient.newBuilder()
+        .connectTimeout(CONNECT_TIMEOUT)
+        .version(HttpClient.Version.HTTP_2)
+        .followRedirects(HttpClient.Redirect.NORMAL)
+        .build();
   }
 
   /** Latest client-level rate-limit snapshot, or {@code null} if no request has succeeded yet. */
@@ -90,19 +103,33 @@ public <T> CompletableFuture<T> executeAsync(RequestSpec spec, Class<T> response
     URI uri = buildUri(spec);
     HttpRequest request = buildRequest(uri);
 
+    // ADR-007: acquire returns a CompletableFuture instead of parking the caller's thread.
+    // When permits are available the future is already completed (fast path) and thenCompose
+    // runs synchronously; when the pool is exhausted the future completes later, on the
+    // thread that calls release() — the caller's thread is never blocked here.
+    return concurrencyPermits.acquire().thenCompose(unused -> dispatch(uri, request, responseType));
+  }
+
+  private <T> CompletableFuture<T> dispatch(URI uri, HttpRequest request, Class<T> responseType) {
+    CompletableFuture<HttpResponse<byte[]>> sendFuture;
     try {
-      concurrencyPermits.acquire();
-    } catch (InterruptedException e) {
-      Thread.currentThread().interrupt();
+      sendFuture = httpClient.sendAsync(request, BodyHandlers.ofByteArray());
+    } catch (Throwable t) {
+      // sendAsync threw synchronously (e.g. malformed request, internal NPE, OOM).
+      // The future never formed, so whenComplete will not fire — release the permit
+      // here to prevent a permanent leak that would degrade the pool to deadlock.
+      concurrencyPermits.release();
+      if (t instanceof Error err) {
+        throw err;
+      }
       return CompletableFuture.failedFuture(
           new NetworkError(
-              "Interrupted while waiting for a concurrency permit",
+              "Request to " + uri + " failed before dispatch: " + t.getMessage(),
               new ErrorContext(null, uri.toString(), null),
-              e));
+              t));
     }
 
-    return httpClient
-        .sendAsync(request, BodyHandlers.ofByteArray())
+    return sendFuture
         .whenComplete((r, t) -> concurrencyPermits.release())
         .handle(
             (response, error) -> {

src/test/java/com/marketdata/sdk/internal/http/AsyncSemaphoreTest.java

@@ -0,0 +1,153 @@
+package com.marketdata.sdk.internal.http;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import org.junit.jupiter.api.Test;
+
+class AsyncSemaphoreTest {
+
+  // ---------- fast path ----------
+
+  @Test
+  void acquireReturnsCompletedFutureWhenPermitsAvailable() {
+    AsyncSemaphore sem = new AsyncSemaphore(3);
+
+    CompletableFuture<Void> a = sem.acquire();
+    CompletableFuture<Void> b = sem.acquire();
+    CompletableFuture<Void> c = sem.acquire();
+
+    assertThat(a).isCompleted();
+    assertThat(b).isCompleted();
+    assertThat(c).isCompleted();
+    assertThat(sem.availablePermits()).isZero();
+    assertThat(sem.queueLength()).isZero();
+  }
+
+  // ---------- slow path ----------
+
+  @Test
+  void acquireReturnsPendingFutureWhenPoolExhausted() {
+    AsyncSemaphore sem = new AsyncSemaphore(2);
+    sem.acquire();
+    sem.acquire();
+
+    CompletableFuture<Void> waiter = sem.acquire();
+
+    assertThat(waiter).isNotCompleted();
+    assertThat(sem.availablePermits()).isZero();
+    assertThat(sem.queueLength()).isOne();
+  }
+
+  @Test
+  void releaseTransfersPermitDirectlyToFirstWaiter() {
+    AsyncSemaphore sem = new AsyncSemaphore(1);
+    sem.acquire(); // pool empty
+
+    CompletableFuture<Void> w1 = sem.acquire();
+    CompletableFuture<Void> w2 = sem.acquire();
+
+    sem.release();
+
+    // The permit goes from the in-flight caller straight to w1 — never re-counted.
+    assertThat(w1).isCompleted();
+    assertThat(w2).isNotCompleted();
+    assertThat(sem.availablePermits()).isZero();
+    assertThat(sem.queueLength()).isOne();
+
+    sem.release();
+
+    assertThat(w2).isCompleted();
+    assertThat(sem.availablePermits()).isZero();
+    assertThat(sem.queueLength()).isZero();
+  }
+
+  @Test
+  void releaseWithNoWaitersIncrementsCounter() {
+    AsyncSemaphore sem = new AsyncSemaphore(2);
+    sem.acquire();
+    sem.acquire();
+
+    sem.release();
+    assertThat(sem.availablePermits()).isOne();
+
+    sem.release();
+    assertThat(sem.availablePermits()).isEqualTo(2);
+  }
+
+  // ---------- cancellation ----------
+
+  @Test
+  void cancelledWaiterIsSkippedOnRelease() {
+    AsyncSemaphore sem = new AsyncSemaphore(1);
+    sem.acquire(); // pool empty
+
+    CompletableFuture<Void> cancelled = sem.acquire();
+    CompletableFuture<Void> alive = sem.acquire();
+    cancelled.cancel(false);
+
+    sem.release();
+
+    // The cancelled waiter is skipped; the next live one gets the permit.
+    assertThat(alive).isCompleted();
+    assertThat(sem.queueLength()).isZero();
+    assertThat(sem.availablePermits()).isZero();
+  }
+
+  @Test
+  void releaseWhenAllWaitersCancelledFallsBackToCounter() {
+    AsyncSemaphore sem = new AsyncSemaphore(1);
+    sem.acquire();
+
+    sem.acquire().cancel(false);
+    sem.acquire().cancel(false);
+
+    sem.release();
+
+    // No live waiter — the permit goes back to the pool.
+    assertThat(sem.availablePermits()).isOne();
+    assertThat(sem.queueLength()).isZero();
+  }
+
+  // ---------- ordering ----------
+
+  @Test
+  void waitersAreServedFifo() {
+    AsyncSemaphore sem = new AsyncSemaphore(0);
+    List<Integer> completionOrder = new ArrayList<>();
+
+    for (int i = 0; i < 10; i++) {
+      int id = i;
+      sem.acquire().thenRun(() -> completionOrder.add(id));
+    }
+
+    for (int i = 0; i < 10; i++) {
+      sem.release();
+    }
+
+    assertThat(completionOrder).containsExactly(0, 1, 2, 3, 4, 5, 6, 7, 8, 9);
+  }
+
+  // ---------- argument validation ----------
+
+  @Test
+  void rejectsNegativeInitialPermits() {
+    assertThatThrownBy(() -> new AsyncSemaphore(-1))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("permits");
+  }
+
+  @Test
+  void zeroInitialPermitsIsValidAndForcesSlowPath() {
+    AsyncSemaphore sem = new AsyncSemaphore(0);
+
+    CompletableFuture<Void> w = sem.acquire();
+    assertThat(w).isNotCompleted();
+
+    sem.release();
+    assertThat(w).isCompleted();
+  }
+}