diff --git a/httpclient5/src/main/java/org/apache/hc/client5/http/impl/nio/DefaultAsyncClientConnectionOperator.java b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/nio/DefaultAsyncClientConnectionOperator.java index e0f0d5050d..bb72be9f34 100644 --- a/httpclient5/src/main/java/org/apache/hc/client5/http/impl/nio/DefaultAsyncClientConnectionOperator.java +++ b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/nio/DefaultAsyncClientConnectionOperator.java @@ -58,6 +58,7 @@ import org.apache.hc.core5.reactor.IOSession; import org.apache.hc.core5.reactor.ssl.TransportSecurityLayer; import org.apache.hc.core5.util.Args; +import org.apache.hc.core5.util.TimeValue; import org.apache.hc.core5.util.Timeout; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -83,9 +84,21 @@ protected DefaultAsyncClientConnectionOperator( final Lookup tlsStrategyLookup, final SchemePortResolver schemePortResolver, final DnsResolver dnsResolver) { + this(tlsStrategyLookup, schemePortResolver, dnsResolver, false, null); + } + + /** + * @since 5.7 + */ + protected DefaultAsyncClientConnectionOperator( + final Lookup tlsStrategyLookup, + final SchemePortResolver schemePortResolver, + final DnsResolver dnsResolver, + final boolean happyEyeballsEnabled, + final TimeValue connectionAttemptDelay) { this.tlsStrategyLookup = Args.notNull(tlsStrategyLookup, "TLS strategy lookup"); this.schemePortResolver = schemePortResolver != null ? schemePortResolver : DefaultSchemePortResolver.INSTANCE; - this.sessionRequester = new MultihomeIOSessionRequester(dnsResolver); + this.sessionRequester = new MultihomeIOSessionRequester(dnsResolver, happyEyeballsEnabled, connectionAttemptDelay); } @Override diff --git a/httpclient5/src/main/java/org/apache/hc/client5/http/impl/nio/MultihomeConnectionInitiator.java b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/nio/MultihomeConnectionInitiator.java index 2e8e00bbf0..04ff6e09e1 100644 --- a/httpclient5/src/main/java/org/apache/hc/client5/http/impl/nio/MultihomeConnectionInitiator.java +++ b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/nio/MultihomeConnectionInitiator.java @@ -38,6 +38,7 @@ import org.apache.hc.core5.reactor.ConnectionInitiator; import org.apache.hc.core5.reactor.IOSession; import org.apache.hc.core5.util.Args; +import org.apache.hc.core5.util.TimeValue; import org.apache.hc.core5.util.Timeout; /** @@ -58,6 +59,36 @@ public MultihomeConnectionInitiator( this.sessionRequester = new MultihomeIOSessionRequester(dnsResolver); } + /** + * Creates a multi-home connection initiator that, for hosts resolving to more than one + * address, optionally races connection attempts per RFC 8305 (Happy Eyeballs v2) instead + * of trying each address sequentially: the first address is tried immediately, and each + * subsequent one after {@code connectionAttemptDelay} unless an outstanding attempt fails + * first. The first successful connection wins; every other attempt is cancelled, and a + * connection that completes after a winner was already decided is closed rather than kept + * open. When {@code happyEyeballsEnabled} is {@code false}, behavior is identical to + * {@link #MultihomeConnectionInitiator(ConnectionInitiator, DnsResolver)}. + * + * @param connectionInitiator the connection initiator. + * @param dnsResolver the DNS resolver to use. + * @param happyEyeballsEnabled whether to race connection attempts instead of trying them + * sequentially. + * @param connectionAttemptDelay delay between the start of successive connection attempts + * (RFC 8305 Section 8 recommends 250ms; must be at least + * 10ms). Ignored when {@code happyEyeballsEnabled} is + * {@code false}. May be {@code null} to use the RFC 8305 + * recommended default. + * @since 5.7 + */ + public MultihomeConnectionInitiator( + final ConnectionInitiator connectionInitiator, + final DnsResolver dnsResolver, + final boolean happyEyeballsEnabled, + final TimeValue connectionAttemptDelay) { + this.connectionInitiator = Args.notNull(connectionInitiator, "Connection initiator"); + this.sessionRequester = new MultihomeIOSessionRequester(dnsResolver, happyEyeballsEnabled, connectionAttemptDelay); + } + @Override public Future connect( final NamedEndpoint remoteEndpoint, diff --git a/httpclient5/src/main/java/org/apache/hc/client5/http/impl/nio/MultihomeIOSessionRequester.java b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/nio/MultihomeIOSessionRequester.java index 4b8172c4ce..58c80cb13a 100644 --- a/httpclient5/src/main/java/org/apache/hc/client5/http/impl/nio/MultihomeIOSessionRequester.java +++ b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/nio/MultihomeIOSessionRequester.java @@ -32,18 +32,29 @@ import java.net.InetSocketAddress; import java.net.SocketAddress; import java.net.UnknownHostException; +import java.util.ArrayList; import java.util.List; +import java.util.concurrent.CancellationException; +import java.util.concurrent.Executors; import java.util.concurrent.Future; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import org.apache.hc.client5.http.ConnectExceptionSupport; import org.apache.hc.client5.http.DnsResolver; import org.apache.hc.client5.http.SystemDefaultDnsResolver; +import org.apache.hc.core5.concurrent.Cancellable; import org.apache.hc.core5.concurrent.ComplexFuture; +import org.apache.hc.core5.concurrent.DefaultThreadFactory; import org.apache.hc.core5.concurrent.FutureCallback; +import org.apache.hc.core5.io.CloseMode; import org.apache.hc.core5.net.NamedEndpoint; import org.apache.hc.core5.reactor.ConnectionInitiator; import org.apache.hc.core5.reactor.IOSession; +import org.apache.hc.core5.util.Args; +import org.apache.hc.core5.util.TimeValue; import org.apache.hc.core5.util.Timeout; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -51,10 +62,65 @@ final class MultihomeIOSessionRequester { private static final Logger LOG = LoggerFactory.getLogger(MultihomeIOSessionRequester.class); + + /** + * RFC 8305, Section 8: "When a reasonable value is unknown, implementations SHOULD use a + * Connection Attempt Delay of 250 milliseconds". + */ + static final TimeValue DEFAULT_CONNECTION_ATTEMPT_DELAY = TimeValue.ofMilliseconds(250); + + /** + * RFC 8305, Section 8: "this value SHOULD NOT be below 10 ms". + */ + private static final long MIN_CONNECTION_ATTEMPT_DELAY_MS = 10; + + /** + * Lazily initialized, process-wide, single daemon-thread scheduler used solely to pace + * staggered connection attempts (RFC 8305 connection racing). Shared across all instances + * so that enabling this feature never adds a thread per connection manager, let alone per + * connection or per request. + */ + private static final class LazyStaggerScheduler { + static final ScheduledExecutorService INSTANCE = Executors.newSingleThreadScheduledExecutor( + new DefaultThreadFactory("Connection-attempt-stagger", true)); + } + private final DnsResolver dnsResolver; + private final boolean happyEyeballsEnabled; + private final TimeValue connectionAttemptDelay; + private final ScheduledExecutorService staggerScheduler; MultihomeIOSessionRequester(final DnsResolver dnsResolver) { + this(dnsResolver, false, DEFAULT_CONNECTION_ATTEMPT_DELAY); + } + + MultihomeIOSessionRequester( + final DnsResolver dnsResolver, + final boolean happyEyeballsEnabled, + final TimeValue connectionAttemptDelay) { + this(dnsResolver, happyEyeballsEnabled, connectionAttemptDelay, null); + } + + // Package-private: allows unit tests to supply a deterministic, synchronous scheduler + // instead of the real, time-based one. + MultihomeIOSessionRequester( + final DnsResolver dnsResolver, + final boolean happyEyeballsEnabled, + final TimeValue connectionAttemptDelay, + final ScheduledExecutorService staggerScheduler) { this.dnsResolver = dnsResolver != null ? dnsResolver : SystemDefaultDnsResolver.INSTANCE; + this.happyEyeballsEnabled = happyEyeballsEnabled; + final TimeValue delay = connectionAttemptDelay != null ? connectionAttemptDelay : DEFAULT_CONNECTION_ATTEMPT_DELAY; + if (happyEyeballsEnabled) { + Args.check(delay.toMilliseconds() >= MIN_CONNECTION_ATTEMPT_DELAY_MS, + "Connection attempt delay must be at least %s ms (RFC 8305, Section 8)", MIN_CONNECTION_ATTEMPT_DELAY_MS); + } + this.connectionAttemptDelay = delay; + this.staggerScheduler = staggerScheduler; + } + + private ScheduledExecutorService staggerScheduler() { + return staggerScheduler != null ? staggerScheduler : LazyStaggerScheduler.INSTANCE; } public Future connect( @@ -123,6 +189,13 @@ public void cancelled() { LOG.debug("{} resolved to {}", remoteEndpoint.getHostName(), remoteAddresses); } + if (happyEyeballsEnabled && remoteAddresses.size() > 1) { + final Race race = new Race(connectionInitiator, remoteEndpoint, remoteAddresses, localAddress, + connectTimeout, attachment, future); + race.start(); + return future; + } + final Runnable runnable = new Runnable() { private final AtomicInteger attempt = new AtomicInteger(0); @@ -207,4 +280,293 @@ public Future connect( return connect(connectionInitiator, remoteEndpoint, null, localAddress, connectTimeout, attachment, callback); } + /** + * RFC 8305 connection racing: the first address is tried immediately; each subsequent + * address is tried after {@link #connectionAttemptDelay} unless an outstanding attempt + * fails first, in which case the next address is tried immediately. The first successful + * connection wins: all other outstanding attempts are cancelled, and any connection that + * completes successfully after a winner has already been decided is closed rather than + * leaked. Exactly one terminal callback (success or aggregated failure) ever reaches + * {@code future}. + */ + private final class Race implements Cancellable { + + private final ConnectionInitiator connectionInitiator; + private final NamedEndpoint remoteEndpoint; + private final List remoteAddresses; + private final SocketAddress localAddress; + private final Timeout connectTimeout; + private final Object attachment; + private final ComplexFuture future; + + private final Object lock = new Object(); + private final List> inFlight = new ArrayList<>(); + private int nextIndex; + private int outstanding; + private boolean decided; + private boolean cancelled; + private ScheduledFuture pendingStagger; + private long lastAttemptStartNanos = -1L; + + Race( + final ConnectionInitiator connectionInitiator, + final NamedEndpoint remoteEndpoint, + final List remoteAddresses, + final SocketAddress localAddress, + final Timeout connectTimeout, + final Object attachment, + final ComplexFuture future) { + this.connectionInitiator = connectionInitiator; + this.remoteEndpoint = remoteEndpoint; + this.remoteAddresses = remoteAddresses; + this.localAddress = localAddress; + this.connectTimeout = connectTimeout; + this.attachment = attachment; + this.future = future; + // A single Cancellable dependency covers every attempt this race ever starts, + // including ones not yet started when the caller cancels (see #cancel()). + this.future.setDependency(this); + } + + void start() { + synchronized (lock) { + scheduleNextPromptlyLocked(); + } + } + + /** + * Starts the next address now if at least the RFC 8305 Section 5 minimum + * inter-attempt spacing (10ms) has passed since the last attempt was started, + * including one that just failed or was unexpectedly cancelled; otherwise schedules + * it for exactly the remaining part of that minimum. Used whenever an attempt should + * advance as promptly as RFC 8305 allows, rather than waiting out the full + * {@link #connectionAttemptDelay}. + * + *

Must be called while holding {@code lock}.

+ */ + private void scheduleNextPromptlyLocked() { + if (decided || cancelled || nextIndex >= remoteAddresses.size()) { + return; + } + if (lastAttemptStartNanos < 0) { + startNextLocked(); + return; + } + final long elapsedMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - lastAttemptStartNanos); + if (elapsedMs >= MIN_CONNECTION_ATTEMPT_DELAY_MS) { + startNextLocked(); + } else { + pendingStagger = staggerScheduler().schedule( + this::onStaggerElapsed, MIN_CONNECTION_ATTEMPT_DELAY_MS - elapsedMs, TimeUnit.MILLISECONDS); + } + } + + // Must be called while holding `lock`. + private void startNextLocked() { + if (decided || cancelled || nextIndex >= remoteAddresses.size()) { + return; + } + final int index = nextIndex++; + final InetSocketAddress remoteAddress = remoteAddresses.get(index); + outstanding++; + lastAttemptStartNanos = System.nanoTime(); + + if (LOG.isDebugEnabled()) { + LOG.debug("{}:{} racing attempt {} {}->{} ({})", + remoteEndpoint.getHostName(), remoteEndpoint.getPort(), index, localAddress, remoteAddress, connectTimeout); + } + + final Future attemptFuture = connectionInitiator.connect( + remoteEndpoint, + remoteAddress, + localAddress, + connectTimeout, + attachment, + new FutureCallback() { + + @Override + public void completed(final IOSession session) { + onAttemptCompleted(remoteAddress, session); + } + + @Override + public void failed(final Exception cause) { + onAttemptFailed(remoteAddress, cause); + } + + @Override + public void cancelled() { + onAttemptCancelled(remoteAddress); + } + + }); + inFlight.add(attemptFuture); + + // pendingStagger != null here means the attempt above already completed + // synchronously and its own failure/cancellation handler already scheduled the + // next advance (possibly at the shorter minimum spacing): don't overwrite it. + if (!decided && !cancelled && pendingStagger == null && nextIndex < remoteAddresses.size()) { + pendingStagger = staggerScheduler().schedule( + this::onStaggerElapsed, connectionAttemptDelay.getDuration(), connectionAttemptDelay.getTimeUnit()); + } + } + + private void onStaggerElapsed() { + synchronized (lock) { + pendingStagger = null; + startNextLocked(); + } + } + + private void onAttemptCompleted(final InetSocketAddress remoteAddress, final IOSession session) { + final boolean won; + final List> losers; + synchronized (lock) { + outstanding--; + won = !decided && !cancelled; + if (won) { + decided = true; + cancelPendingStaggerLocked(); + losers = new ArrayList<>(inFlight); + } else { + losers = null; + } + } + if (won) { + if (LOG.isDebugEnabled()) { + LOG.debug("{}:{} race won by {}->{} as {}", + remoteEndpoint.getHostName(), remoteEndpoint.getPort(), localAddress, remoteAddress, session.getId()); + } + for (final Future loser : losers) { + loser.cancel(true); + } + // future may already have been cancelled directly by the caller (not through + // this Race's own #cancel()) in the brief window between the checks above and + // here: ComplexFuture#completed() returns false in that case, and the session + // must still be closed rather than silently leaked. + if (!future.completed(session)) { + if (LOG.isDebugEnabled()) { + LOG.debug("{}:{} race won by {}->{} as {} but the future was already done; closing", + remoteEndpoint.getHostName(), remoteEndpoint.getPort(), localAddress, remoteAddress, session.getId()); + } + session.close(CloseMode.IMMEDIATE); + } + } else { + // A duplicate winner (concurrent success) or a late completion after the + // race was already decided/cancelled: never leak the socket. + if (LOG.isDebugEnabled()) { + LOG.debug("{}:{} discarding late successful connection {}->{} ({})", + remoteEndpoint.getHostName(), remoteEndpoint.getPort(), localAddress, remoteAddress, session.getId()); + } + session.close(CloseMode.IMMEDIATE); + } + } + + private void onAttemptFailed(final InetSocketAddress remoteAddress, final Exception cause) { + boolean advance = false; + boolean last = false; + synchronized (lock) { + outstanding--; + if (decided || cancelled) { + return; + } + cancelPendingStaggerLocked(); + if (nextIndex < remoteAddresses.size()) { + advance = true; + } else if (outstanding == 0) { + decided = true; + last = true; + } + if (advance) { + // RFC 8305 Section 5: advance to the next address as promptly as possible + // after a failure, never waiting out the full connection attempt delay, + // but still respecting the minimum inter-attempt spacing. + scheduleNextPromptlyLocked(); + } + } + if (last) { + if (LOG.isDebugEnabled()) { + LOG.debug("{}:{} connection to {} failed ({}); all addresses exhausted", + remoteEndpoint.getHostName(), remoteEndpoint.getPort(), remoteAddress, cause.getClass()); + } + finish(cause); + } + } + + /** + * An attempt may be cancelled by something other than this race itself deciding a + * winner or being cancelled as a whole, for example an external close of the + * connection manager. Treat it exactly like a failure so the overall future never + * stays pending forever: advance to the next address if one remains, or finish with a + * failure if none does. + */ + private void onAttemptCancelled(final InetSocketAddress remoteAddress) { + boolean advance = false; + boolean last = false; + synchronized (lock) { + outstanding--; + if (decided || cancelled) { + // Expected: either this race decided a winner and cancelled every other + // attempt itself, or the whole race was cancelled from the outside. + return; + } + cancelPendingStaggerLocked(); + if (nextIndex < remoteAddresses.size()) { + advance = true; + } else if (outstanding == 0) { + decided = true; + last = true; + } + if (advance) { + scheduleNextPromptlyLocked(); + } + } + if (last) { + if (LOG.isDebugEnabled()) { + LOG.debug("{}:{} connection attempt to {} was unexpectedly cancelled; all addresses exhausted", + remoteEndpoint.getHostName(), remoteEndpoint.getPort(), remoteAddress); + } + finish(new CancellationException( + "Connection attempt to " + remoteAddress + " was cancelled")); + } + } + + private void finish(final Exception cause) { + if (cause instanceof IOException) { + final InetAddress[] addresses = remoteAddresses.stream() + .map(InetSocketAddress::getAddress) + .toArray(InetAddress[]::new); + future.failed(ConnectExceptionSupport.enhance((IOException) cause, remoteEndpoint, addresses)); + } else { + future.failed(cause); + } + } + + @Override + public boolean cancel() { + final List> toCancel; + synchronized (lock) { + if (decided || cancelled) { + return false; + } + cancelled = true; + cancelPendingStaggerLocked(); + toCancel = new ArrayList<>(inFlight); + } + for (final Future attemptFuture : toCancel) { + attemptFuture.cancel(true); + } + return true; + } + + // Must be called while holding `lock`. + private void cancelPendingStaggerLocked() { + if (pendingStagger != null) { + pendingStagger.cancel(false); + pendingStagger = null; + } + } + + } + } diff --git a/httpclient5/src/main/java/org/apache/hc/client5/http/impl/nio/PoolingAsyncClientConnectionManagerBuilder.java b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/nio/PoolingAsyncClientConnectionManagerBuilder.java index a4e1b8fe48..d35bf8d609 100644 --- a/httpclient5/src/main/java/org/apache/hc/client5/http/impl/nio/PoolingAsyncClientConnectionManagerBuilder.java +++ b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/nio/PoolingAsyncClientConnectionManagerBuilder.java @@ -57,6 +57,8 @@ public class PoolingAsyncClientConnectionManagerBuilder { private TlsStrategy tlsStrategy; private SchemePortResolver schemePortResolver; private DnsResolver dnsResolver; + private boolean happyEyeballsEnabled; + private TimeValue connectionAttemptDelay; private PoolConcurrencyPolicy poolConcurrencyPolicy; private PoolReusePolicy poolReusePolicy; @@ -99,6 +101,37 @@ public final PoolingAsyncClientConnectionManagerBuilder setDnsResolver(final Dns return this; } + /** + * Enables RFC 8305 (Happy Eyeballs v2) connection racing for hosts that resolve to more + * than one address: the first address is tried immediately, and each subsequent one after + * {@link #setConnectionAttemptDelay(TimeValue)} unless an outstanding attempt fails first. + * The first successful connection wins; every other attempt is cancelled. Disabled + * ({@code false}) by default, in which case addresses continue to be tried sequentially. + * + * @param happyEyeballsEnabled whether to race connection attempts. + * @return this instance. + * @since 5.7 + */ + public final PoolingAsyncClientConnectionManagerBuilder setHappyEyeballsEnabled(final boolean happyEyeballsEnabled) { + this.happyEyeballsEnabled = happyEyeballsEnabled; + return this; + } + + /** + * Sets the delay between the start of successive connection attempts when + * {@link #setHappyEyeballsEnabled(boolean)} is enabled. RFC 8305 Section 8 recommends + * 250ms and requires at least 10ms. Has no effect unless Happy Eyeballs racing is enabled. + * + * @param connectionAttemptDelay the delay, or {@code null} to use the RFC 8305 + * recommended default of 250ms. + * @return this instance. + * @since 5.7 + */ + public final PoolingAsyncClientConnectionManagerBuilder setConnectionAttemptDelay(final TimeValue connectionAttemptDelay) { + this.connectionAttemptDelay = connectionAttemptDelay; + return this; + } + /** * Sets {@link SchemePortResolver} instance. * @@ -267,12 +300,27 @@ protected AsyncClientConnectionOperator createConnectionOperator( final TlsStrategy tlsStrategy, final SchemePortResolver schemePortResolver, final DnsResolver dnsResolver) { + return createConnectionOperator(tlsStrategy, schemePortResolver, dnsResolver, false, null); + } + + /** + * @since 5.7 + */ + @Internal + protected AsyncClientConnectionOperator createConnectionOperator( + final TlsStrategy tlsStrategy, + final SchemePortResolver schemePortResolver, + final DnsResolver dnsResolver, + final boolean happyEyeballsEnabled, + final TimeValue connectionAttemptDelay) { return new DefaultAsyncClientConnectionOperator( RegistryBuilder.create() .register(URIScheme.HTTPS.getId(), tlsStrategy) .build(), schemePortResolver, - dnsResolver); + dnsResolver, + happyEyeballsEnabled, + connectionAttemptDelay); } public PoolingAsyncClientConnectionManager build() { @@ -287,7 +335,8 @@ public PoolingAsyncClientConnectionManager build() { } } final PoolingAsyncClientConnectionManager poolingmgr = new PoolingAsyncClientConnectionManager( - createConnectionOperator(tlsStrategyCopy, schemePortResolver, dnsResolver), + createConnectionOperator(tlsStrategyCopy, schemePortResolver, dnsResolver, + happyEyeballsEnabled, connectionAttemptDelay), poolConcurrencyPolicy, poolReusePolicy, null, diff --git a/httpclient5/src/test/java/org/apache/hc/client5/http/impl/nio/MultihomeIOSessionRequesterTest.java b/httpclient5/src/test/java/org/apache/hc/client5/http/impl/nio/MultihomeIOSessionRequesterTest.java index e2ec5696b8..3997b2ee21 100644 --- a/httpclient5/src/test/java/org/apache/hc/client5/http/impl/nio/MultihomeIOSessionRequesterTest.java +++ b/httpclient5/src/test/java/org/apache/hc/client5/http/impl/nio/MultihomeIOSessionRequesterTest.java @@ -28,29 +28,45 @@ package org.apache.hc.client5.http.impl.nio; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertInstanceOf; import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.api.Assertions.fail; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.eq; import java.io.IOException; import java.net.InetAddress; import java.net.InetSocketAddress; +import java.util.ArrayDeque; +import java.util.ArrayList; import java.util.Arrays; +import java.util.Deque; import java.util.List; +import java.util.concurrent.CancellationException; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; import org.apache.hc.client5.http.DnsResolver; import org.apache.hc.core5.concurrent.FutureCallback; +import org.apache.hc.core5.io.CloseMode; import org.apache.hc.core5.net.NamedEndpoint; import org.apache.hc.core5.reactor.ConnectionInitiator; import org.apache.hc.core5.reactor.IOSession; +import org.apache.hc.core5.util.TimeValue; import org.apache.hc.core5.util.Timeout; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; import org.mockito.Mockito; class MultihomeIOSessionRequesterTest { @@ -158,4 +174,515 @@ void testConnectSuccessfulAfterRetries() throws Exception { fail("Did not expect an ExecutionException", ex); } } + + // ------------------------------------------------------------------------- + // RFC 8305 connection racing (Happy Eyeballs v2) + // ------------------------------------------------------------------------- + + /** Captures scheduled stagger tasks instead of running them on a timer, for determinism. */ + private static final class FakeScheduler { + final ScheduledExecutorService mock = Mockito.mock(ScheduledExecutorService.class); + final Deque pending = new ArrayDeque<>(); + + @SuppressWarnings("unchecked") + FakeScheduler() { + Mockito.when(mock.schedule(any(Runnable.class), anyLong(), any(TimeUnit.class))) + .thenAnswer(invocation -> { + pending.add(invocation.getArgument(0)); + return Mockito.mock(ScheduledFuture.class); + }); + } + + /** Runs the next pending stagger task, as if its delay had elapsed. */ + void fireNext() { + final Runnable task = pending.poll(); + assertNotNull(task, "no stagger task was scheduled"); + task.run(); + } + + int pendingCount() { + return pending.size(); + } + } + + private static List addresses(final int count) throws Exception { + final List out = new ArrayList<>(count); + for (int i = 1; i <= count; i++) { + out.add(new InetSocketAddress(InetAddress.getByAddress(new byte[]{10, 0, 0, (byte) i}), 8080)); + } + return out; + } + + @SuppressWarnings("unchecked") + private static FutureCallback[] captureCallbacks(final ConnectionInitiator connectionInitiator, final int expected) { + final ArgumentCaptor captor = ArgumentCaptor.forClass(FutureCallback.class); + Mockito.verify(connectionInitiator, Mockito.times(expected)) + .connect(any(), any(), any(), any(), any(), captor.capture()); + return captor.getAllValues().toArray(new FutureCallback[0]); + } + + @Test + void happyEyeballs_firstAddressSucceedsImmediately_noStaggerNeeded() throws Exception { + final List remoteAddresses = addresses(2); + Mockito.when(namedEndpoint.getHostName()).thenReturn("somehost"); + Mockito.when(namedEndpoint.getPort()).thenReturn(8080); + Mockito.when(dnsResolver.resolve("somehost", 8080)).thenReturn(remoteAddresses); + + final IOSession winnerSession = Mockito.mock(IOSession.class); + final FakeScheduler scheduler = new FakeScheduler(); + Mockito.when(connectionInitiator.connect(any(), any(), any(), any(), any(), any())) + .thenAnswer(invocation -> { + final FutureCallback cb = invocation.getArgument(5); + cb.completed(winnerSession); + return CompletableFuture.completedFuture(winnerSession); + }); + + final MultihomeIOSessionRequester requester = + new MultihomeIOSessionRequester(dnsResolver, true, TimeValue.ofMilliseconds(250), scheduler.mock); + final Future future = requester.connect( + connectionInitiator, namedEndpoint, null, Timeout.ofSeconds(5), null, null); + + assertTrue(future.isDone()); + assertEquals(winnerSession, future.get()); + // Only the first address was ever attempted; the second was never needed. + Mockito.verify(connectionInitiator, Mockito.times(1)).connect(any(), any(), any(), any(), any(), any()); + Mockito.verify(winnerSession, Mockito.never()).close(any(CloseMode.class)); + } + + @Test + void happyEyeballs_firstAddressStalls_secondWinsAfterStagger() throws Exception { + final List remoteAddresses = addresses(2); + Mockito.when(namedEndpoint.getHostName()).thenReturn("somehost"); + Mockito.when(namedEndpoint.getPort()).thenReturn(8080); + Mockito.when(dnsResolver.resolve("somehost", 8080)).thenReturn(remoteAddresses); + + final IOSession winnerSession = Mockito.mock(IOSession.class); + final FakeScheduler scheduler = new FakeScheduler(); + final Future firstAttempt = Mockito.mock(Future.class); + Mockito.when(connectionInitiator.connect(any(), any(), any(), any(), any(), any())) + .thenAnswer(invocation -> firstAttempt) // 1st call: stalls, no callback invoked yet. + .thenAnswer(invocation -> { + final FutureCallback cb = invocation.getArgument(5); + cb.completed(winnerSession); + return CompletableFuture.completedFuture(winnerSession); + }); + + final MultihomeIOSessionRequester requester = + new MultihomeIOSessionRequester(dnsResolver, true, TimeValue.ofMilliseconds(250), scheduler.mock); + final Future future = requester.connect( + connectionInitiator, namedEndpoint, null, Timeout.ofSeconds(5), null, null); + + assertFalse(future.isDone(), "must not complete before the stalled first attempt or the stagger fires"); + assertEquals(1, scheduler.pendingCount()); + + scheduler.fireNext(); // simulate the 250ms stagger elapsing + Mockito.verify(connectionInitiator, Mockito.times(2)).connect(any(), any(), any(), any(), any(), any()); + assertTrue(future.isDone()); + assertEquals(winnerSession, future.get()); + + // The still-stalled first attempt must have been cancelled, not left to leak. + Mockito.verify(firstAttempt).cancel(true); + } + + @Test + void happyEyeballs_firstAddressFailsFast_advancesPromptlyAtMinimumSpacingNotFullStagger() throws Exception { + final List remoteAddresses = addresses(2); + Mockito.when(namedEndpoint.getHostName()).thenReturn("somehost"); + Mockito.when(namedEndpoint.getPort()).thenReturn(8080); + Mockito.when(dnsResolver.resolve("somehost", 8080)).thenReturn(remoteAddresses); + + final IOSession winnerSession = Mockito.mock(IOSession.class); + final FakeScheduler scheduler = new FakeScheduler(); + Mockito.when(connectionInitiator.connect(any(), any(), any(), any(), any(), any())) + .thenAnswer(invocation -> { + final FutureCallback cb = invocation.getArgument(5); + cb.failed(new IOException("refused")); + return CompletableFuture.completedFuture(null); + }) + .thenAnswer(invocation -> { + final FutureCallback cb = invocation.getArgument(5); + cb.completed(winnerSession); + return CompletableFuture.completedFuture(winnerSession); + }); + + final MultihomeIOSessionRequester requester = + new MultihomeIOSessionRequester(dnsResolver, true, TimeValue.ofMilliseconds(250), scheduler.mock); + final Future future = requester.connect( + connectionInitiator, namedEndpoint, null, Timeout.ofSeconds(5), null, null); + + // The 1st address failed immediately: the 2nd must be scheduled at the RFC 8305 + // Section 5 minimum inter-attempt spacing (10ms), not the full 250ms stagger delay. + assertFalse(future.isDone()); + final ArgumentCaptor delayCaptor = ArgumentCaptor.forClass(Long.class); + Mockito.verify(scheduler.mock).schedule(any(Runnable.class), delayCaptor.capture(), eq(TimeUnit.MILLISECONDS)); + assertTrue(delayCaptor.getValue() <= 10L && delayCaptor.getValue() > 0L, + "must advance at the minimum spacing (<=10ms), not the full stagger delay: was " + delayCaptor.getValue()); + + scheduler.fireNext(); + assertTrue(future.isDone()); + assertEquals(winnerSession, future.get()); + Mockito.verify(connectionInitiator, Mockito.times(2)).connect(any(), any(), any(), any(), any(), any()); + } + + @Test + void happyEyeballs_bothAttemptsSucceed_loserSessionIsClosedNotLeaked() throws Exception { + final List remoteAddresses = addresses(2); + Mockito.when(namedEndpoint.getHostName()).thenReturn("somehost"); + Mockito.when(namedEndpoint.getPort()).thenReturn(8080); + Mockito.when(dnsResolver.resolve("somehost", 8080)).thenReturn(remoteAddresses); + + final IOSession secondSession = Mockito.mock(IOSession.class); + final Future firstAttempt = Mockito.mock(Future.class); + final FakeScheduler scheduler = new FakeScheduler(); + Mockito.when(connectionInitiator.connect(any(), any(), any(), any(), any(), any())) + .thenAnswer(invocation -> firstAttempt) // stalls until completed manually below + .thenAnswer(invocation -> { + final FutureCallback cb = invocation.getArgument(5); + cb.completed(secondSession); + return CompletableFuture.completedFuture(secondSession); + }); + + final MultihomeIOSessionRequester requester = + new MultihomeIOSessionRequester(dnsResolver, true, TimeValue.ofMilliseconds(250), scheduler.mock); + final Future future = requester.connect( + connectionInitiator, namedEndpoint, null, Timeout.ofSeconds(5), null, null); + + // The first attempt stalls, so the stagger fires and starts the second attempt, which + // wins the race immediately. + assertFalse(future.isDone()); + scheduler.fireNext(); + assertTrue(future.isDone()); + assertEquals(secondSession, future.get()); + Mockito.verify(secondSession, Mockito.never()).close(any(CloseMode.class)); + + // The first address now completes late, after the race was already decided: its + // session must be discarded rather than leaked or allowed to override the winner. + final FutureCallback[] allCallbacks = captureCallbacks(connectionInitiator, 2); + final IOSession lateSession = Mockito.mock(IOSession.class); + allCallbacks[0].completed(lateSession); + + Mockito.verify(lateSession).close(CloseMode.IMMEDIATE); + assertEquals(secondSession, future.get(), "the already-decided winner must not change"); + } + + @Test + void happyEyeballs_allAddressesFail_aggregatedFailure() throws Exception { + final List remoteAddresses = addresses(3); + Mockito.when(namedEndpoint.getHostName()).thenReturn("somehost"); + Mockito.when(namedEndpoint.getPort()).thenReturn(8080); + Mockito.when(dnsResolver.resolve("somehost", 8080)).thenReturn(remoteAddresses); + + final FakeScheduler scheduler = new FakeScheduler(); + Mockito.when(connectionInitiator.connect(any(), any(), any(), any(), any(), any())) + .thenAnswer(invocation -> { + final FutureCallback cb = invocation.getArgument(5); + cb.failed(new IOException("refused")); + return CompletableFuture.completedFuture(null); + }); + + final MultihomeIOSessionRequester requester = + new MultihomeIOSessionRequester(dnsResolver, true, TimeValue.ofMilliseconds(250), scheduler.mock); + final Future future = requester.connect( + connectionInitiator, namedEndpoint, null, Timeout.ofSeconds(5), null, null); + + // All 3 addresses fail fast, one after another; each advance is scheduled at the + // RFC 8305 minimum inter-attempt spacing (10ms) rather than happening synchronously. + assertFalse(future.isDone()); + scheduler.fireNext(); + assertFalse(future.isDone()); + scheduler.fireNext(); + assertTrue(future.isDone()); + Mockito.verify(connectionInitiator, Mockito.times(3)).connect(any(), any(), any(), any(), any(), any()); + try { + future.get(); + fail("Expected ExecutionException"); + } catch (final ExecutionException ex) { + assertInstanceOf(IOException.class, ex.getCause()); + } + } + + @Test + void happyEyeballs_cancellationBeforeAnyAttemptResolves_cancelsOutstandingAndPendingStagger() throws Exception { + final List remoteAddresses = addresses(2); + Mockito.when(namedEndpoint.getHostName()).thenReturn("somehost"); + Mockito.when(namedEndpoint.getPort()).thenReturn(8080); + Mockito.when(dnsResolver.resolve("somehost", 8080)).thenReturn(remoteAddresses); + + final FakeScheduler scheduler = new FakeScheduler(); + final Future firstAttempt = Mockito.mock(Future.class); + Mockito.when(connectionInitiator.connect(any(), any(), any(), any(), any(), any())) + .thenAnswer(invocation -> firstAttempt); + + final MultihomeIOSessionRequester requester = + new MultihomeIOSessionRequester(dnsResolver, true, TimeValue.ofMilliseconds(250), scheduler.mock); + final Future future = requester.connect( + connectionInitiator, namedEndpoint, null, Timeout.ofSeconds(5), null, null); + + assertFalse(future.isDone()); + assertEquals(1, scheduler.pendingCount()); + + future.cancel(true); + + Mockito.verify(firstAttempt).cancel(true); + assertTrue(future.isCancelled()); + } + + @Test + void happyEyeballs_lateCompletionAfterCancellation_closesSessionWithoutReachingFuture() throws Exception { + final List remoteAddresses = addresses(2); + Mockito.when(namedEndpoint.getHostName()).thenReturn("somehost"); + Mockito.when(namedEndpoint.getPort()).thenReturn(8080); + Mockito.when(dnsResolver.resolve("somehost", 8080)).thenReturn(remoteAddresses); + + final FakeScheduler scheduler = new FakeScheduler(); + Mockito.when(connectionInitiator.connect(any(), any(), any(), any(), any(), any())) + .thenAnswer(invocation -> Mockito.mock(Future.class)); + + final MultihomeIOSessionRequester requester = + new MultihomeIOSessionRequester(dnsResolver, true, TimeValue.ofMilliseconds(250), scheduler.mock); + final Future future = requester.connect( + connectionInitiator, namedEndpoint, null, Timeout.ofSeconds(5), null, null); + future.cancel(true); + + final FutureCallback[] callbacks = captureCallbacks(connectionInitiator, 1); + final IOSession lateSession = Mockito.mock(IOSession.class); + callbacks[0].completed(lateSession); + + Mockito.verify(lateSession).close(CloseMode.IMMEDIATE); + assertTrue(future.isCancelled()); + } + + @Test + void happyEyeballs_singleAddress_doesNotRace() throws Exception { + final List remoteAddresses = addresses(1); + Mockito.when(namedEndpoint.getHostName()).thenReturn("somehost"); + Mockito.when(namedEndpoint.getPort()).thenReturn(8080); + Mockito.when(dnsResolver.resolve("somehost", 8080)).thenReturn(remoteAddresses); + + final IOSession session = Mockito.mock(IOSession.class); + final FakeScheduler scheduler = new FakeScheduler(); + Mockito.when(connectionInitiator.connect(any(), any(), any(), any(), any(), any())) + .thenAnswer(invocation -> { + final FutureCallback cb = invocation.getArgument(5); + cb.completed(session); + return CompletableFuture.completedFuture(session); + }); + + final MultihomeIOSessionRequester requester = + new MultihomeIOSessionRequester(dnsResolver, true, TimeValue.ofMilliseconds(250), scheduler.mock); + final Future future = requester.connect( + connectionInitiator, namedEndpoint, null, Timeout.ofSeconds(5), null, null); + + assertTrue(future.isDone()); + assertEquals(session, future.get()); + assertEquals(0, scheduler.pendingCount(), "a single address must never schedule a stagger"); + } + + @Test + void happyEyeballs_disabledByDefault_usesSequentialPathEvenWithMultipleAddresses() throws Exception { + final List remoteAddresses = addresses(2); + Mockito.when(namedEndpoint.getHostName()).thenReturn("somehost"); + Mockito.when(namedEndpoint.getPort()).thenReturn(8080); + Mockito.when(dnsResolver.resolve("somehost", 8080)).thenReturn(remoteAddresses); + + final IOSession session = Mockito.mock(IOSession.class); + Mockito.when(connectionInitiator.connect(any(), any(), any(), any(), any(), any())) + .thenAnswer(invocation -> { + final FutureCallback cb = invocation.getArgument(5); + cb.completed(session); + return CompletableFuture.completedFuture(session); + }); + + // Explicit happyEyeballsEnabled=false, as opposed to the default single-arg constructor + // already exercised by the other tests in this class. + final MultihomeIOSessionRequester requester = + new MultihomeIOSessionRequester(dnsResolver, false, TimeValue.ofMilliseconds(250)); + final Future future = requester.connect( + connectionInitiator, namedEndpoint, null, Timeout.ofSeconds(5), null, null); + + assertTrue(future.isDone()); + assertEquals(session, future.get()); + Mockito.verify(connectionInitiator, Mockito.times(1)).connect(any(), any(), any(), any(), any(), any()); + } + + @Test + void happyEyeballs_respectsConfiguredDelay() throws Exception { + final List remoteAddresses = addresses(2); + Mockito.when(namedEndpoint.getHostName()).thenReturn("somehost"); + Mockito.when(namedEndpoint.getPort()).thenReturn(8080); + Mockito.when(dnsResolver.resolve("somehost", 8080)).thenReturn(remoteAddresses); + + Mockito.when(connectionInitiator.connect(any(), any(), any(), any(), any(), any())) + .thenAnswer(invocation -> Mockito.mock(Future.class)); + + final ScheduledExecutorService schedulerMock = Mockito.mock(ScheduledExecutorService.class); + Mockito.when(schedulerMock.schedule(any(Runnable.class), anyLong(), any(TimeUnit.class))) + .thenReturn(Mockito.mock(ScheduledFuture.class)); + + final MultihomeIOSessionRequester requester = new MultihomeIOSessionRequester( + dnsResolver, true, TimeValue.ofMilliseconds(777), schedulerMock); + requester.connect(connectionInitiator, namedEndpoint, null, Timeout.ofSeconds(5), null, null); + + Mockito.verify(schedulerMock).schedule(any(Runnable.class), org.mockito.ArgumentMatchers.eq(777L), org.mockito.ArgumentMatchers.eq(TimeUnit.MILLISECONDS)); + } + + @Test + void happyEyeballs_rejectsDelayBelowRfc8305Minimum() { + assertThrows(IllegalArgumentException.class, + () -> new MultihomeIOSessionRequester(dnsResolver, true, TimeValue.ofMilliseconds(5))); + } + + @Test + void happyEyeballs_unexpectedCancellationOfNonLastAttempt_advancesToNextAddress() throws Exception { + final List remoteAddresses = addresses(2); + Mockito.when(namedEndpoint.getHostName()).thenReturn("somehost"); + Mockito.when(namedEndpoint.getPort()).thenReturn(8080); + Mockito.when(dnsResolver.resolve("somehost", 8080)).thenReturn(remoteAddresses); + + final IOSession winnerSession = Mockito.mock(IOSession.class); + final FakeScheduler scheduler = new FakeScheduler(); + Mockito.when(connectionInitiator.connect(any(), any(), any(), any(), any(), any())) + .thenAnswer(invocation -> { + // Cancelled by something other than this race itself (e.g. an external + // close), not a failure and not our own decided/cancel() path. + final FutureCallback cb = invocation.getArgument(5); + cb.cancelled(); + return CompletableFuture.completedFuture(null); + }) + .thenAnswer(invocation -> { + final FutureCallback cb = invocation.getArgument(5); + cb.completed(winnerSession); + return CompletableFuture.completedFuture(winnerSession); + }); + + final MultihomeIOSessionRequester requester = + new MultihomeIOSessionRequester(dnsResolver, true, TimeValue.ofMilliseconds(250), scheduler.mock); + final Future future = requester.connect( + connectionInitiator, namedEndpoint, null, Timeout.ofSeconds(5), null, null); + + assertFalse(future.isDone()); + scheduler.fireNext(); // the minimum-spacing advance after the unexpected cancellation + assertTrue(future.isDone()); + assertEquals(winnerSession, future.get()); + } + + @Test + void happyEyeballs_unexpectedCancellationOfLastAttempt_failsRatherThanHangingForever() throws Exception { + final List remoteAddresses = addresses(1); + Mockito.when(namedEndpoint.getHostName()).thenReturn("somehost"); + Mockito.when(namedEndpoint.getPort()).thenReturn(8080); + Mockito.when(dnsResolver.resolve("somehost", 8080)).thenReturn(remoteAddresses); + + // A single address racing alone still goes through the Race (not the sequential path) + // when forced via the 4-arg constructor is not applicable here; use 2 addresses with + // the 2nd address also cancelled, so the race is driven to full exhaustion. + final List twoAddresses = addresses(2); + Mockito.when(dnsResolver.resolve("somehost", 8080)).thenReturn(twoAddresses); + + final FakeScheduler scheduler = new FakeScheduler(); + Mockito.when(connectionInitiator.connect(any(), any(), any(), any(), any(), any())) + .thenAnswer(invocation -> { + final FutureCallback cb = invocation.getArgument(5); + cb.cancelled(); + return CompletableFuture.completedFuture(null); + }); + + final MultihomeIOSessionRequester requester = + new MultihomeIOSessionRequester(dnsResolver, true, TimeValue.ofMilliseconds(250), scheduler.mock); + final Future future = requester.connect( + connectionInitiator, namedEndpoint, null, Timeout.ofSeconds(5), null, null); + + assertFalse(future.isDone()); + scheduler.fireNext(); + assertTrue(future.isDone(), "the future must not be left pending forever after every attempt is cancelled"); + assertTrue(future.isCancelled() || future.isDone()); + try { + future.get(); + fail("Expected an ExecutionException rather than an indefinitely pending future"); + } catch (final ExecutionException ex) { + assertInstanceOf(CancellationException.class, ex.getCause()); + } + } + + /** + * Reproduces the race between {@code Race#onAttemptCompleted} picking a winner and the + * caller cancelling the outer {@code Future} directly (not through the Race's own + * {@code cancel()}): {@code ComplexFuture#completed(T)} can then return {@code false} + * even though the Race's own bookkeeping says it won, and the winning session must still + * be closed rather than silently leaked. This genuinely requires two threads contending + * on the same narrow window inside {@code ComplexFuture#cancel}, so the test runs many + * iterations under real contention and asserts the invariant (no leaked session, exactly + * one terminal outcome) holds on every one, rather than depending on hitting the window + * a specific number of times. + */ + @Test + void happyEyeballs_concurrentExternalCancelDuringWinnerSelection_neverLeaksSession() throws Exception { + for (int i = 0; i < 500; i++) { + // 2 addresses so this goes through the Race (not the sequential) path; the 2nd + // attempt is left permanently stalled behind a FakeScheduler so no real timer + // ever fires during the test. + final List remoteAddresses = addresses(2); + final DnsResolver localDnsResolver = Mockito.mock(DnsResolver.class); + final ConnectionInitiator localConnectionInitiator = Mockito.mock(ConnectionInitiator.class); + final NamedEndpoint localEndpoint = Mockito.mock(NamedEndpoint.class); + Mockito.when(localEndpoint.getHostName()).thenReturn("somehost"); + Mockito.when(localEndpoint.getPort()).thenReturn(8080); + Mockito.when(localDnsResolver.resolve("somehost", 8080)).thenReturn(remoteAddresses); + + final IOSession session = Mockito.mock(IOSession.class); + final AtomicReference> attemptCallback = new AtomicReference<>(); + final FakeScheduler localScheduler = new FakeScheduler(); + Mockito.when(localConnectionInitiator.connect(any(), any(), any(), any(), any(), any())) + .thenAnswer(invocation -> { + attemptCallback.set(invocation.getArgument(5)); + return Mockito.mock(Future.class); + }); + + final MultihomeIOSessionRequester requester = new MultihomeIOSessionRequester( + localDnsResolver, true, TimeValue.ofMilliseconds(250), localScheduler.mock); + final Future future = requester.connect( + localConnectionInitiator, localEndpoint, null, Timeout.ofSeconds(5), null, null); + + final CountDownLatch ready = new CountDownLatch(2); + final CountDownLatch go = new CountDownLatch(1); + final Thread canceller = new Thread(() -> { + ready.countDown(); + await(go); + future.cancel(true); + }); + final Thread completer = new Thread(() -> { + ready.countDown(); + await(go); + attemptCallback.get().completed(session); + }); + canceller.start(); + completer.start(); + ready.await(); + go.countDown(); + canceller.join(5000); + completer.join(5000); + + // Exactly one terminal outcome, and the session is never left open. + assertTrue(future.isDone()); + boolean sessionDelivered = false; + try { + sessionDelivered = future.get() == session; + } catch (final CancellationException | ExecutionException ex) { + sessionDelivered = false; + } + if (!sessionDelivered) { + Mockito.verify(session, Mockito.atLeastOnce()).close(CloseMode.IMMEDIATE); + } else { + Mockito.verify(session, Mockito.never()).close(any(CloseMode.class)); + } + } + } + + private static void await(final CountDownLatch latch) { + try { + latch.await(); + } catch (final InterruptedException ex) { + Thread.currentThread().interrupt(); + } + } }