When applying pendingAcquireTimeout, check for current pending Borrowers

reactor-pool/src/main/java/reactor/pool/AbstractPool.java

@@ -122,6 +122,18 @@ public boolean isInactiveForMoreThan(Duration duration) {
 
 	// == common methods to interact with idle/pending queues ==
 
+	void applyPendingAcquireTimeout(Borrower<POOLABLE> borrower) {
+		boolean noIdle = idleSize() == 0;
+		boolean noPermits = poolConfig.allocationStrategy().estimatePermitCount() == 0;
+
+		if (noIdle && noPermits) {
+			borrower.pendingAcquireStart = clock.millis();
+			if (!borrower.pendingAcquireTimeout.isZero()) {
+				borrower.timeoutTask = config().pendingAcquireTimer().apply(borrower, borrower.pendingAcquireTimeout);
+			}
+		}
+	}
+
 	abstract boolean elementOffer(POOLABLE element); //used in tests
 
 	/**
@@ -423,16 +435,8 @@ public void run() {
 		public void request(long n) {
 			if (Operators.validate(n)) {
 				//start the countdown
+				pool.applyPendingAcquireTimeout(this);
 
-				boolean noIdle = pool.idleSize() == 0;
-				boolean noPermits = pool.poolConfig.allocationStrategy().estimatePermitCount() == 0;
-
-				if (noIdle && noPermits) {
-					pendingAcquireStart = pool.clock.millis();
-					if (!pendingAcquireTimeout.isZero()) {
-						timeoutTask = this.pool.config().pendingAcquireTimer().apply(this, pendingAcquireTimeout);
-					}
-				}
 				//doAcquire should interrupt the countdown if there is either an available
 				//resource or the pool can allocate one
 				pool.doAcquire(this);

reactor-pool/src/main/java/reactor/pool/SimpleDequePool.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2018-2023 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2018-2024 VMware Inc. or its affiliates, 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.
@@ -508,6 +508,11 @@ public int pendingAcquireSize() {
 		return pendingSize;
 	}
 
+	@Override
+	void applyPendingAcquireTimeout(Borrower<POOLABLE> borrower) {
+		// Pending acquire timeout is applied when pendingOffer() is invoked
+	}
+
 	@Override
 	boolean elementOffer(POOLABLE element) {
 		@SuppressWarnings("unchecked")
@@ -571,7 +576,6 @@ final void maybeRecycleAndDrain(QueuePooledRef<POOLABLE> poolSlot, CoreSubscribe
 	 * @param pending a new {@link reactor.pool.AbstractPool.Borrower} to add to the queue and later either serve or consider pending
 	 */
 	void pendingOffer(Borrower<POOLABLE> pending) {
-		int maxPending = poolConfig.maxPending();
 		ConcurrentLinkedDeque<Borrower<POOLABLE>> pendingQueue = this.pending;
 		if (pendingQueue == TERMINATED) {
 			return;
@@ -581,6 +585,17 @@ void pendingOffer(Borrower<POOLABLE> pending) {
 			postOffer = PENDING_SIZE.incrementAndGet(this);
 		}
 
+		int idle = idleSize;
+		int estimatePermitCount = poolConfig.allocationStrategy().estimatePermitCount();
+
+		if (idle + estimatePermitCount < postOffer) {
+			pending.pendingAcquireStart = clock.millis();
+			if (!pending.pendingAcquireTimeout.isZero()) {
+				pending.timeoutTask = config().pendingAcquireTimer().apply(pending, pending.pendingAcquireTimeout);
+			}
+		}
+
+		int maxPending = poolConfig.maxPending();
 		if (WIP.getAndIncrement(this) == 0) {
 			if (maxPending >= 0 && postOffer > maxPending && idleSize == 0 && poolConfig.allocationStrategy().estimatePermitCount() == 0) {
 				//fail fast. differentiate slightly special case of maxPending == 0

reactor-pool/src/test/java/reactor/pool/CommonPoolTest.java

@@ -31,6 +31,8 @@
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -1117,6 +1119,31 @@ void discardCloseableWhenCloseFailureLogs(PoolStyle configAdjuster) {
 		}
 	}
 
+	@ParameterizedTestWithName
+	@MethodSource("allPools")
+	void pendingTimeout(PoolStyle configAdjuster) throws Exception {
+		PoolBuilder<String, ?> builder = PoolBuilder
+				.from(Mono.just("pendingTimeout"))
+				.sizeBetween(0, 1)
+				.maxPendingAcquire(10);
+		AbstractPool<String> pool = configAdjuster.apply(builder);
+
+		CountDownLatch latch = new CountDownLatch(3);
+		ExecutorService executorService = Executors.newFixedThreadPool(20);
+		CompletableFuture<?>[] completableFutures = new CompletableFuture<?>[4];
+		for (int i = 0; i < completableFutures.length; i++) {
+			completableFutures[i] = CompletableFuture.runAsync(
+					() -> pool.acquire(Duration.ofMillis(10))
+							.doOnError(t -> latch.countDown())
+							.onErrorResume(PoolAcquireTimeoutException.class, t -> Mono.empty())
+							.block(),
+					executorService);
+		}
+
+		Mono.fromCompletionStage(CompletableFuture.allOf(completableFutures)).block(Duration.ofSeconds(5));
+		assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue();
+	}
+
 	@ParameterizedTestWithName
 	@MethodSource("allPools")
 	void pendingTimeoutNotImpactedByLongAllocation(PoolStyle configAdjuster) {