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 <[email protected]>

Author: violetagg

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

@@ -655,8 +655,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();
+				}
 			}
 		}
 
@@ -760,14 +763,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;
@@ -855,6 +862,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) {
@@ -866,7 +883,8 @@ void stopPendingCountdown(boolean success) {
 
 				pendingAcquireStart = 0;
 			}
-			timeoutTask.dispose();
+			Disposable task = TIMEOUT_TASK.getAndSet(this, TIMEOUT_STOPPED);
+			task.dispose();
 		}
 	}