Fix race condition in pending acquire timer management (#4073)

There was a synchronisation issue between pendingOffer and drainLoop
where a timer could be set after stopPendingCountdown was called,
resulting in a timer that would never be stopped and would incorrectly
fire even though the borrower had already been served.

Signed-off-by: Violeta Georgieva <696661+violetagg@users.noreply.github.com>

Author: violetagg

reactor-netty-http/src/main/java/reactor/netty/http/client/Http2Pool.java

@@ -656,8 +656,11 @@ void pendingOffer(Borrower borrower) {
 		int permits = poolConfig.allocationStrategy().estimatePermitCount();
 		if (permits + estimateStreamsCount < postOffer) {
 			borrower.pendingAcquireStart = clock.millis();
-			if (!borrower.acquireTimeout.isZero()) {
-				borrower.timeoutTask = poolConfig.pendingAcquireTimer().apply(borrower, borrower.acquireTimeout);
+			if (!borrower.acquireTimeout.isZero() && borrower.timeoutTask == Borrower.TIMEOUT_DISPOSED) {
+				Disposable task = poolConfig.pendingAcquireTimer().apply(borrower, borrower.acquireTimeout);
+				if (!borrower.setTimeoutTask(task)) {
+					task.dispose();
+				}
 			}
 		}
 
@@ -763,14 +766,18 @@ void scheduleEviction() {
 	static final class Borrower extends AtomicBoolean implements Scannable, Subscription, Runnable {
 
 		static final Disposable TIMEOUT_DISPOSED = Disposables.disposed();
+		static final Disposable TIMEOUT_STOPPED = Disposables.disposed();
 
 		final Duration acquireTimeout;
 		final CoreSubscriber<? super Http2PooledRef> actual;
 		final Http2Pool pool;
 
 		long pendingAcquireStart;
 
-		Disposable timeoutTask;
+		volatile Disposable timeoutTask;
+		@SuppressWarnings("rawtypes")
+		static final AtomicReferenceFieldUpdater<Borrower, Disposable> TIMEOUT_TASK =
+				AtomicReferenceFieldUpdater.newUpdater(Borrower.class, Disposable.class, "timeoutTask");
 
 		Borrower(CoreSubscriber<? super Http2PooledRef> actual, Http2Pool pool, Duration acquireTimeout) {
 			this.acquireTimeout = acquireTimeout;
@@ -851,6 +858,16 @@ void fail(Throwable error) {
 			}
 		}
 
+		/**
+		 * Atomically set the timeout task if not already stopped.
+		 *
+		 * @return true if the task was set, false if countdown was already stopped
+		 * @see #stopPendingCountdown(boolean)
+		 */
+		boolean setTimeoutTask(Disposable task) {
+			return TIMEOUT_TASK.compareAndSet(this, TIMEOUT_DISPOSED, task);
+		}
+
 		void stopPendingCountdown(boolean success) {
 			if (pendingAcquireStart > 0) {
 				if (success) {
@@ -862,7 +879,8 @@ void stopPendingCountdown(boolean success) {
 
 				pendingAcquireStart = 0;
 			}
-			timeoutTask.dispose();
+			Disposable task = TIMEOUT_TASK.getAndSet(this, TIMEOUT_STOPPED);
+			task.dispose();
 		}
 	}