fix(bulkhead): release permits granted to abandoned waiters

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

@@ -51,10 +51,14 @@ public BulkheadConfig<R> getConfig() {
 
   @Override
   public void acquirePermit() throws InterruptedException {
+    CompletableFuture<Void> future = acquirePermitAsync();
     try {
-      acquirePermitAsync().get();
+      future.get();
     } catch (CancellationException | ExecutionException ignore) {
       // Not possible since the future will always be completed with null
+    } catch (InterruptedException e) {
+      abandonWaiter(future);
+      throw e;
     }
   }
 
@@ -77,10 +81,23 @@ public boolean tryAcquirePermit(Duration maxWaitTime) throws InterruptedExceptio
       future.get(maxWaitTime.toNanos(), TimeUnit.NANOSECONDS);
       return true;
     } catch (CancellationException | ExecutionException | TimeoutException e) {
+      abandonWaiter(future);
       return false;
+    } catch (InterruptedException e) {
+      abandonWaiter(future);
+      throw e;
     }
   }
 
+  /**
+   * Removes an abandoned {@code waiter} from the queue. If a permit was concurrently granted to the waiter by {@link
+   * #releasePermit()}, the permit is released again so that it's not lost.
+   */
+  private void abandonWaiter(CompletableFuture<Void> waiter) {
+    if (!waiter.complete(null))
+      releasePermit();
+  }
+
   /**
    * Returns a CompletableFuture that is completed when a permit is acquired. Externally completing this future will
    * remove the waiter from the bulkhead's internal queue.

core/src/test/java/dev/failsafe/internal/BulkheadImplTest.java

@@ -0,0 +1,67 @@
+/*
+ * Copyright 2024 the original author or authors.
+ *
+ * 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 dev.failsafe.internal;
+
+import dev.failsafe.Bulkhead;
+import org.testng.annotations.Test;
+
+import java.time.Duration;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertTrue;
+
+/**
+ * Tests that a Bulkhead does not lose permits when a timed acquisition gives up.
+ */
+@Test
+public class BulkheadImplTest {
+  private static final Duration SHORT_WAIT = Duration.ofMillis(20);
+
+  public void shouldRetainPermitAfterAcquireTimesOut() throws Throwable {
+    Bulkhead<Object> bulkhead = Bulkhead.builder(1).build();
+    assertTrue(bulkhead.tryAcquirePermit());
+
+    assertFalse(bulkhead.tryAcquirePermit(SHORT_WAIT));
+    bulkhead.releasePermit();
+
+    assertTrue(bulkhead.tryAcquirePermit(), "the released permit must be available again");
+  }
+
+  public void shouldRetainCapacityOverRepeatedAcquireTimeouts() throws Throwable {
+    int maxConcurrency = 4;
+    Bulkhead<Object> bulkhead = Bulkhead.builder(maxConcurrency).build();
+
+    for (int round = 1; round <= 5; round++) {
+      for (int i = 0; i < maxConcurrency; i++)
+        assertTrue(bulkhead.tryAcquirePermit());
+      assertFalse(bulkhead.tryAcquirePermit(SHORT_WAIT));
+      for (int i = 0; i < maxConcurrency; i++)
+        bulkhead.releasePermit();
+
+      assertEquals(capacityOf(bulkhead), maxConcurrency, "capacity after round " + round);
+    }
+  }
+
+  private static int capacityOf(Bulkhead<Object> bulkhead) {
+    int permits = 0;
+    while (bulkhead.tryAcquirePermit())
+      permits++;
+    for (int i = 0; i < permits; i++)
+      bulkhead.releasePermit();
+    return permits;
+  }
+}