Fix bulkhead race condition

Previous implementation made it possible for permit to be acquired, but client code getting `BulkheadFullException` resulting in permit leak.

Author: mar-kolya

core/src/main/java/dev/failsafe/internal/BulkheadExecutor.java

@@ -24,6 +24,7 @@
 import dev.failsafe.spi.Scheduler;
 
 import java.time.Duration;
+import java.util.concurrent.CancellationException;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
@@ -62,22 +63,33 @@ protected CompletableFuture<ExecutionResult<R>> preExecuteAsync(Scheduler schedu
     CompletableFuture<ExecutionResult<R>> promise = new CompletableFuture<>();
     CompletableFuture<Void> acquireFuture = bulkhead.acquirePermitAsync();
     acquireFuture.whenComplete((result, error) -> {
-      // Signal for execution to proceed
-      promise.complete(ExecutionResult.none());
+      if (error instanceof CancellationException) {
+        // Cancellation of acquireFuture future means either cancellation in the scheduler (in which case we probably
+        // do not care about the result too much), or cancellation because we reached maxWaitTime (see below) - in which case
+        // we want to inform the user with BulkheadFullException.
+        promise.complete(ExecutionResult.exception(new BulkheadFullException(bulkhead)));
+      } else {
+        // Signal for execution to proceed
+        promise.complete(ExecutionResult.none());
+      }
     });
 
     if (!promise.isDone()) {
       try {
         // Schedule bulkhead permit timeout
         Future<?> timeoutFuture = scheduler.schedule(() -> {
-          promise.complete(ExecutionResult.exception(new BulkheadFullException(bulkhead)));
+          // Note: we cannot call `promise.complete` here directly. Doing so would result in a following race condition:
+          // * `promise` would be considered failed (i.e. caller would think thar no permit was acquired)
+          // * but some other thread may release permit before we call `acquireFuture.cancel` - resulting in `acquireFuture` being completed
+          // successfully (and permit acquired). But since `promise` is already completed that fact is ignored.
+          // This discrepancy would lead to permits 'leaking'. So instead we make `acquireFuture.whenComplete` be the only way
+          // to complete `promise` and here we just signal to that code that `promise` should return BulkheadFullException.
           acquireFuture.cancel(true);
           return null;
         }, maxWaitTime.toNanos(), TimeUnit.NANOSECONDS);
 
         // Propagate outer cancellations to the promise, bulkhead acquire future, and timeout future
         future.setCancelFn(this, (mayInterrupt, cancelResult) -> {
-          promise.complete(cancelResult);
           acquireFuture.cancel(mayInterrupt);
           timeoutFuture.cancel(mayInterrupt);
         });

core/src/test/java/dev/failsafe/functional/BulkheadTest.java

@@ -25,8 +25,14 @@
 import org.testng.annotations.Test;
 
 import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.testng.Assert.assertTrue;
 
 /**
  * Tests various Bulkhead scenarios.
@@ -96,4 +102,46 @@ public void testMaxWaitTimeExceeded() {
     testRunFailure(Failsafe.with(bulkhead), ctx -> {
     }, BulkheadFullException.class);
   }
+
+  @Test
+  public void testPermitsLeak() throws InterruptedException {
+    // We verify against leak of permits because of a race condition that only happens when maxWaitTime is not zero.
+    Bulkhead<Object> bulkhead = Bulkhead.builder(1).withMaxWaitTime(Duration.ofMillis(1)).build();
+    FailsafeExecutor<Object> failsafe = Failsafe.with(bulkhead);
+
+    AtomicInteger errors = new AtomicInteger();
+    List<Thread> threads = new ArrayList<>();
+    for (int i = 0; i < 30; i++) {
+      threads.add(new Thread(() -> {
+        for (int j = 0; j < 30; j++) {
+          try {
+            failsafe.getStageAsync(() -> {
+              try {
+                Thread.sleep(10);
+              } catch (InterruptedException e) {
+                log(getClass(), "Interrupted sleep", e);
+              }
+              return null;
+            }).join(); // Submit work to the bulkhead
+          } catch (CompletionException e) {
+            errors.incrementAndGet();
+          }
+        }
+      }));
+    }
+
+    // Start and join the threads
+    threads.forEach(Thread::start);
+    for (Thread t : threads) {
+      t.join();
+    }
+
+    // Wait for the executor to finish all work. 250ms is plenty of time to finish the work submitted above.
+    Thread.sleep(250);
+
+    // Make sure this run doesn't fail
+    failsafe.getStageAsync(() -> null).join();
+
+    assertTrue(errors.get() > 0, "Should have some errors because maxWaitTime is very small");
+  }
 }