From 9a052a29ef3741818e5cbdc705d2d5753487167f Mon Sep 17 00:00:00 2001 From: Arturo Bernal Date: Tue, 10 Nov 2026 21:32:00 +0100 Subject: [PATCH] Race connection attempts across resolved addresses for the Classic client Replaces the one-thread-per-address, sleep-based stagger with a single dispatcher that starts the first address immediately and waits up to the connection attempt delay for an outstanding attempt to finish before starting the next; on a failure it advances immediately, at the RFC 8305 minimum spacing (10ms), instead of waiting out the full delay. No worker thread ever sleeps a delay itself, bounding concurrency to the number of resolved addresses actually in flight. Socket registration goes through a small atomic registry instead of a bare reference: deciding a winner and a worker registering its socket can happen at the same instant, and whichever runs second closes the socket, so nothing registered around that moment survives open regardless of timing. shutdownNow() alone cannot abort a blocking Socket#connect; closing the socket is what actually does that, on every exit path (winner found, every address failed, or interrupted) via one cleanup call in a finally block. An interruption during the minimum-spacing wait is now propagated instead of swallowed, aborting the race immediately like interruption anywhere else in it. Racing is skipped in favor of the existing sequential path when a specific, non-ephemeral local port is requested, since binding several attempts to it concurrently would trade a real connect failure for a spurious bind conflict. Public, blocking connect() behavior, TLS, SOCKS and local binding are otherwise unchanged. --- .../DefaultHttpClientConnectionOperator.java | 78 ++++ .../impl/io/HappyEyeballsSocketConnector.java | 293 +++++++++++++++ ...ingHttpClientConnectionManagerBuilder.java | 38 +- ...faultHttpClientConnectionOperatorTest.java | 194 ++++++++++ .../io/HappyEyeballsSocketConnectorTest.java | 340 ++++++++++++++++++ 5 files changed, 942 insertions(+), 1 deletion(-) create mode 100644 httpclient5/src/main/java/org/apache/hc/client5/http/impl/io/HappyEyeballsSocketConnector.java create mode 100644 httpclient5/src/test/java/org/apache/hc/client5/http/impl/io/DefaultHttpClientConnectionOperatorTest.java create mode 100644 httpclient5/src/test/java/org/apache/hc/client5/http/impl/io/HappyEyeballsSocketConnectorTest.java diff --git a/httpclient5/src/main/java/org/apache/hc/client5/http/impl/io/DefaultHttpClientConnectionOperator.java b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/io/DefaultHttpClientConnectionOperator.java index a5f616865f..2517e9b44e 100644 --- a/httpclient5/src/main/java/org/apache/hc/client5/http/impl/io/DefaultHttpClientConnectionOperator.java +++ b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/io/DefaultHttpClientConnectionOperator.java @@ -89,11 +89,24 @@ public class DefaultHttpClientConnectionOperator implements HttpClientConnection static final DetachedSocketFactory PLAIN_SOCKET_FACTORY = socksProxy -> socksProxy == null ? new Socket() : new Socket(socksProxy); + /** + * 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; + private final DetachedSocketFactory detachedSocketFactory; private final UnixDomainSocketFactory unixDomainSocketFactory = UnixDomainSocketFactory.getSocketFactory(); private final Lookup tlsSocketStrategyLookup; private final SchemePortResolver schemePortResolver; private final DnsResolver dnsResolver; + private final boolean happyEyeballsEnabled; + private final TimeValue connectionAttemptDelay; /** * @deprecated Provided for backward compatibility @@ -115,6 +128,19 @@ public DefaultHttpClientConnectionOperator( final SchemePortResolver schemePortResolver, final DnsResolver dnsResolver, final Lookup tlsSocketStrategyLookup) { + this(detachedSocketFactory, schemePortResolver, dnsResolver, tlsSocketStrategyLookup, false, null); + } + + /** + * @since 5.7 + */ + public DefaultHttpClientConnectionOperator( + final DetachedSocketFactory detachedSocketFactory, + final SchemePortResolver schemePortResolver, + final DnsResolver dnsResolver, + final Lookup tlsSocketStrategyLookup, + final boolean happyEyeballsEnabled, + final TimeValue connectionAttemptDelay) { super(); this.detachedSocketFactory = Args.notNull(detachedSocketFactory, "Plain socket factory"); this.tlsSocketStrategyLookup = Args.notNull(tlsSocketStrategyLookup, "Socket factory registry"); @@ -122,6 +148,13 @@ public DefaultHttpClientConnectionOperator( DefaultSchemePortResolver.INSTANCE; 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; } /** @@ -200,6 +233,51 @@ public void connect( final int port = this.schemePortResolver.resolve(endpointHost.getSchemeName(), endpointHost); remoteAddresses = this.dnsResolver.resolve(endpointHost.getHostName(), port); } + + // A specific, non-ephemeral local port can only ever be bound by one socket at a + // time; racing would bind several attempts to it concurrently, and the OS would + // reject all but one with a bind conflict instead of a real connect failure. Fall + // back to the existing sequential path in that case, exactly as before this feature, + // since only one attempt is ever in flight there. + final boolean canRace = localAddress == null || localAddress.getPort() == 0; + if (happyEyeballsEnabled && canRace && remoteAddresses.size() > 1) { + onBeforeSocketConnect(context, endpointHost); + if (LOG.isDebugEnabled()) { + LOG.debug("{} racing connection attempts across {} ({})", endpointHost, remoteAddresses, connectTimeout); + } + final HappyEyeballsSocketConnector connector = new HappyEyeballsSocketConnector(connectionAttemptDelay); + final Socket socket = connector.connect(endpointHost, remoteAddresses, (remoteAddress, socketConsumer) -> { + final Socket attemptSocket = detachedSocketFactory.create(endpointHost.getSchemeName(), socksProxy); + socketConsumer.accept(attemptSocket); + try { + if (localAddress != null) { + attemptSocket.bind(localAddress); + } + configureSocket(attemptSocket, socketConfig, true); + attemptSocket.connect(remoteAddress, TimeValue.isPositive(connectTimeout) ? connectTimeout.toMillisecondsIntBound() : 0); + return attemptSocket; + } catch (final IOException | RuntimeException ex) { + Closer.closeQuietly(attemptSocket); + throw ex; + } + }); + try { + conn.bind(socket); + onAfterSocketConnect(context, endpointHost); + if (LOG.isDebugEnabled()) { + LOG.debug("{} {} connected {}->{}", ConnPoolSupport.getId(conn), endpointHost, conn.getLocalAddress(), conn.getRemoteAddress()); + } + final TlsSocketStrategy tlsSocketStrategy = tlsSocketStrategyLookup != null ? tlsSocketStrategyLookup.lookup(endpointHost.getSchemeName()) : null; + if (tlsSocketStrategy != null) { + upgradeToTls(conn, endpointHost, endpointName, connectTimeout, attachment, context, tlsSocketStrategy, socket); + } + } catch (final IOException | RuntimeException ex) { + Closer.closeQuietly(socket); + throw ex; + } + return; + } + for (int i = 0; i < remoteAddresses.size(); i++) { final InetSocketAddress remoteAddress = remoteAddresses.get(i); final boolean last = i == remoteAddresses.size() - 1; diff --git a/httpclient5/src/main/java/org/apache/hc/client5/http/impl/io/HappyEyeballsSocketConnector.java b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/io/HappyEyeballsSocketConnector.java new file mode 100644 index 0000000000..ffbc0cf640 --- /dev/null +++ b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/io/HappyEyeballsSocketConnector.java @@ -0,0 +1,293 @@ +/* + * ==================================================================== + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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. + * ==================================================================== + * + * This software consists of voluntary contributions made by many + * individuals on behalf of the Apache Software Foundation. For more + * information on the Apache Software Foundation, please see + * . + * + */ +package org.apache.hc.client5.http.impl.io; + +import java.io.IOException; +import java.io.InterruptedIOException; +import java.net.InetSocketAddress; +import java.net.Socket; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorCompletionService; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Consumer; + +import org.apache.hc.client5.http.ConnectExceptionSupport; +import org.apache.hc.core5.concurrent.DefaultThreadFactory; +import org.apache.hc.core5.io.Closer; +import org.apache.hc.core5.net.NamedEndpoint; +import org.apache.hc.core5.util.TimeValue; + +/** + * Races a blocking connect attempt against each of a host's resolved addresses (RFC 8305, + * Happy Eyeballs v2) for the Classic (blocking) client. A single dispatcher, running on the + * calling thread, starts the first address immediately and each subsequent one on its own + * worker thread, bounded to the number of resolved addresses: it waits up to + * {@link #connectionAttemptDelay} for an outstanding attempt to finish before starting the + * next one, but starts the next one immediately, at the RFC 8305 Section 5 minimum spacing + * (10ms), as soon as an attempt fails. No thread ever sleeps out a delay itself. + * + *

The first attempt to connect successfully wins the race. Because closing a socket is + * the only reliable way to abort a blocking {@link Socket#connect}, every other socket, be it + * already connecting, about to be registered, or not yet created, must end up closed; a + * worker thread registers its socket through a shared, atomic registry rather than a bare + * reference, so a socket created in the brief window around the race being decided is always + * either excluded from cleanup because it does not exist yet, or closed by the registry the + * instant it is registered. {@code ExecutorService#shutdownNow()} only interrupts threads; it + * does not itself abort a blocking connect, so it is never relied on for that.

+ * + * @since 5.7 + */ +final class HappyEyeballsSocketConnector { + + /** + * RFC 8305, Section 5: the minimum spacing between the start of successive connection + * attempts, including one started promptly after an earlier attempt's failure. + */ + private static final long MIN_CONNECTION_ATTEMPT_DELAY_MS = 10; + + /** + * Connects a single socket to {@code remoteAddress}, registering the socket via + * {@code socketConsumer} as soon as it is created and before the (blocking) connect + * call, so that a concurrent winner can close it even while still connecting. + */ + @FunctionalInterface + interface ConnectAttempt { + Socket connect(InetSocketAddress remoteAddress, Consumer socketConsumer) throws IOException; + } + + /** + * Registers each attempt's socket so it can be closed by whichever happens first: this + * race deciding a winner, or the socket being registered. {@link #decide} and + * {@link #register} coordinate purely through {@link #decided} and the per-slot + * {@link AtomicReference}, so neither a socket registered just before {@link #decide} + * runs nor one registered just after is ever left open: {@link #decide} closes whatever + * it finds already registered, and {@link #register} closes its own socket immediately + * if it finds the registry already decided, including re-checking right after + * registering to catch a {@link #decide} that ran in between. + */ + private static final class SocketRegistry { + + private final AtomicReference[] slots; + private final AtomicBoolean decided = new AtomicBoolean(false); + + @SuppressWarnings("unchecked") + SocketRegistry(final int n) { + slots = new AtomicReference[n]; + for (int i = 0; i < n; i++) { + slots[i] = new AtomicReference<>(); + } + } + + void register(final int index, final Socket socket) { + if (decided.get()) { + Closer.closeQuietly(socket); + return; + } + slots[index].set(socket); + if (decided.get() && slots[index].compareAndSet(socket, null)) { + Closer.closeQuietly(socket); + } + } + + /** Closes every registered socket except {@code winner} (may be {@code null}). */ + void decide(final Socket winner) { + decided.set(true); + for (final AtomicReference slot : slots) { + final Socket socket = slot.getAndSet(null); + if (socket != null && socket != winner) { + Closer.closeQuietly(socket); + } + } + } + } + + private final TimeValue connectionAttemptDelay; + + HappyEyeballsSocketConnector(final TimeValue connectionAttemptDelay) { + this.connectionAttemptDelay = connectionAttemptDelay; + } + + Socket connect( + final NamedEndpoint remoteEndpoint, + final List remoteAddresses, + final ConnectAttempt connectAttempt) throws IOException { + final int n = remoteAddresses.size(); + final ExecutorService executor = Executors.newFixedThreadPool( + n, new DefaultThreadFactory("Happy-eyeballs-classic-connect", true)); + final SocketRegistry registry = new SocketRegistry(n); + final List> futures = new ArrayList<>(n); + Socket winner = null; + try { + final ExecutorCompletionService completionService = new ExecutorCompletionService<>(executor); + futures.add(submit(completionService, 0, remoteAddresses, connectAttempt, registry)); + int started = 1; + long lastStartNanos = System.nanoTime(); + IOException lastFailure = null; + + for (int received = 0; received < n && winner == null; ) { + final Future completed; + if (started < n) { + final long waitMs = pendingDelayMs(lastStartNanos, connectionAttemptDelay.toMilliseconds()); + completed = poll(completionService, waitMs); + if (completed == null) { + // Nothing finished within the delay: start the next address now. + futures.add(submit(completionService, started, remoteAddresses, connectAttempt, registry)); + started++; + lastStartNanos = System.nanoTime(); + continue; + } + } else { + completed = take(completionService); + } + received++; + + final Outcome outcome; + try { + outcome = completed.get(); + } catch (final ExecutionException ex) { + final Throwable cause = ex.getCause(); + lastFailure = cause instanceof IOException ? (IOException) cause : new IOException(cause); + continue; + } catch (final InterruptedException ex) { + Thread.currentThread().interrupt(); + throw new InterruptedIOException("Interrupted while racing connection attempts"); + } + if (outcome.socket != null) { + winner = outcome.socket; + for (final Future future : futures) { + future.cancel(true); + } + break; + } + if (outcome.failure != null) { + lastFailure = outcome.failure; + } + if (started < n) { + // Advance as promptly as RFC 8305 allows, but never closer than the + // minimum spacing to the most recently started attempt. + final long waitMs = pendingDelayMs(lastStartNanos, MIN_CONNECTION_ATTEMPT_DELAY_MS); + if (waitMs > 0) { + sleepOrAbort(waitMs); + } + futures.add(submit(completionService, started, remoteAddresses, connectAttempt, registry)); + started++; + lastStartNanos = System.nanoTime(); + } + } + if (winner != null) { + return winner; + } + throw lastFailure != null ? + ConnectExceptionSupport.enhance(lastFailure, remoteEndpoint) : + new IOException("Connection to " + remoteEndpoint + " failed for an unknown reason"); + } finally { + // The single point of truth for cleanup, covering every exit path (winner found, + // every address failed, or this method itself was interrupted): closes every + // socket except the winner (if any), including ones registered concurrently with + // this call. shutdownNow() only interrupts worker threads; it cannot itself abort + // a blocking Socket#connect, so the registry above is what actually does that. + registry.decide(winner); + executor.shutdownNow(); + } + } + + private static long pendingDelayMs(final long lastStartNanos, final long targetDelayMs) { + final long elapsedMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - lastStartNanos); + return Math.max(0, targetDelayMs - elapsedMs); + } + + /** + * Blocks the dispatcher thread itself for a bounded, short minimum-spacing gap (at most + * {@link #MIN_CONNECTION_ATTEMPT_DELAY_MS}). Unlike the per-address stagger this replaces, + * this never delays a worker thread's connect attempt and never exceeds a few + * milliseconds, so it does not amount to the blocking the rest of this class avoids. An + * interruption here must abort racing immediately, exactly like one during {@link #poll} + * or {@link #take}: it is propagated rather than swallowed, so the caller's {@code finally} + * block runs cleanup and no further attempt is started. + */ + private static void sleepOrAbort(final long millis) throws InterruptedIOException { + try { + Thread.sleep(millis); + } catch (final InterruptedException ex) { + Thread.currentThread().interrupt(); + throw new InterruptedIOException("Interrupted while racing connection attempts"); + } + } + + private static Future poll(final ExecutorCompletionService completionService, final long timeoutMs) throws InterruptedIOException { + try { + return completionService.poll(timeoutMs, TimeUnit.MILLISECONDS); + } catch (final InterruptedException ex) { + Thread.currentThread().interrupt(); + throw new InterruptedIOException("Interrupted while racing connection attempts"); + } + } + + private static Future take(final ExecutorCompletionService completionService) throws InterruptedIOException { + try { + return completionService.take(); + } catch (final InterruptedException ex) { + Thread.currentThread().interrupt(); + throw new InterruptedIOException("Interrupted while racing connection attempts"); + } + } + + private static Future submit( + final ExecutorCompletionService completionService, + final int index, + final List remoteAddresses, + final ConnectAttempt connectAttempt, + final SocketRegistry registry) { + final InetSocketAddress remoteAddress = remoteAddresses.get(index); + return completionService.submit(() -> { + try { + final Socket socket = connectAttempt.connect(remoteAddress, socket1 -> registry.register(index, socket1)); + return new Outcome(socket, null); + } catch (final IOException ex) { + return new Outcome(null, ex); + } + }); + } + + private static final class Outcome { + final Socket socket; + final IOException failure; + + Outcome(final Socket socket, final IOException failure) { + this.socket = socket; + this.failure = failure; + } + } + +} diff --git a/httpclient5/src/main/java/org/apache/hc/client5/http/impl/io/PoolingHttpClientConnectionManagerBuilder.java b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/io/PoolingHttpClientConnectionManagerBuilder.java index 5256d32033..796eed53c2 100644 --- a/httpclient5/src/main/java/org/apache/hc/client5/http/impl/io/PoolingHttpClientConnectionManagerBuilder.java +++ b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/io/PoolingHttpClientConnectionManagerBuilder.java @@ -65,6 +65,8 @@ public class PoolingHttpClientConnectionManagerBuilder { private TlsSocketStrategy tlsSocketStrategy; private SchemePortResolver schemePortResolver; private DnsResolver dnsResolver; + private boolean happyEyeballsEnabled; + private TimeValue connectionAttemptDelay; private PoolConcurrencyPolicy poolConcurrencyPolicy; private PoolReusePolicy poolReusePolicy; private Resolver socketConfigResolver; @@ -145,6 +147,38 @@ public final PoolingHttpClientConnectionManagerBuilder setDnsResolver(final DnsR 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, on its own thread, and each + * subsequent one on its own thread after {@link #setConnectionAttemptDelay(TimeValue)} + * unless an outstanding attempt fails first. The first successful connection wins; every + * other attempt's socket is closed. Disabled ({@code false}) by default, in which case + * addresses continue to be tried sequentially on the calling thread. + * + * @param happyEyeballsEnabled whether to race connection attempts. + * @return this instance. + * @since 5.7 + */ + public final PoolingHttpClientConnectionManagerBuilder 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 PoolingHttpClientConnectionManagerBuilder setConnectionAttemptDelay(final TimeValue connectionAttemptDelay) { + this.connectionAttemptDelay = connectionAttemptDelay; + return this; + } + /** * Sets {@link SchemePortResolver} instance. * @@ -346,7 +380,9 @@ protected HttpClientConnectionOperator createConnectionOperator( dnsResolver, RegistryBuilder.create() .register(URIScheme.HTTPS.id, tlsSocketStrategy) - .build()); + .build(), + happyEyeballsEnabled, + connectionAttemptDelay); } public PoolingHttpClientConnectionManager build() { diff --git a/httpclient5/src/test/java/org/apache/hc/client5/http/impl/io/DefaultHttpClientConnectionOperatorTest.java b/httpclient5/src/test/java/org/apache/hc/client5/http/impl/io/DefaultHttpClientConnectionOperatorTest.java new file mode 100644 index 0000000000..95bd1ca6f9 --- /dev/null +++ b/httpclient5/src/test/java/org/apache/hc/client5/http/impl/io/DefaultHttpClientConnectionOperatorTest.java @@ -0,0 +1,194 @@ +/* + * ==================================================================== + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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. + * ==================================================================== + * + * This software consists of voluntary contributions made by many + * individuals on behalf of the Apache Software Foundation. For more + * information on the Apache Software Foundation, please see + * . + * + */ +package org.apache.hc.client5.http.impl.io; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.IOException; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.Proxy; +import java.net.ServerSocket; +import java.net.Socket; +import java.time.Duration; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.atomic.AtomicReference; + +import org.apache.hc.client5.http.DnsResolver; +import org.apache.hc.client5.http.impl.DefaultSchemePortResolver; +import org.apache.hc.client5.http.io.DetachedSocketFactory; +import org.apache.hc.client5.http.ssl.TlsSocketStrategy; +import org.apache.hc.core5.http.HttpHost; +import org.apache.hc.core5.http.config.Lookup; +import org.apache.hc.core5.http.io.SocketConfig; +import org.apache.hc.core5.http.protocol.BasicHttpContext; +import org.apache.hc.core5.util.TimeValue; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; + +class DefaultHttpClientConnectionOperatorTest { + + /** + * Regression test: when {@code localAddress} specifies a non-zero (explicit) local port, + * Happy Eyeballs racing must not bind more than one socket to it concurrently, which the + * OS would reject for all but one attempt as a bind conflict rather than a real connect + * failure. The operator falls back to the existing sequential path in that case, so a + * dead first address followed by a real, reachable second address must still succeed, + * exactly as it would against master without this feature. + */ + @Test + void happyEyeballsWithExplicitLocalPort_fallsBackToSequential_stillSucceeds() throws Exception { + try (ServerSocket server = new ServerSocket(0, 1, InetAddress.getLoopbackAddress()); + ServerSocket deadPortProbe = new ServerSocket(0, 1, InetAddress.getLoopbackAddress())) { + final int listeningPort = server.getLocalPort(); + final int deadPort = deadPortProbe.getLocalPort(); + deadPortProbe.close(); // nothing listens here once closed: instant refusal + + final int explicitLocalPort; + try (ServerSocket localPortProbe = new ServerSocket(0, 1, InetAddress.getLoopbackAddress())) { + explicitLocalPort = localPortProbe.getLocalPort(); + } + + final List remoteAddresses = Arrays.asList( + new InetSocketAddress(InetAddress.getLoopbackAddress(), deadPort), + new InetSocketAddress(InetAddress.getLoopbackAddress(), listeningPort)); + final DnsResolver dnsResolver = Mockito.mock(DnsResolver.class); + Mockito.when(dnsResolver.resolve(Mockito.anyString(), Mockito.anyInt())).thenReturn(remoteAddresses); + + final DefaultHttpClientConnectionOperator operator = new DefaultHttpClientConnectionOperator( + DefaultHttpClientConnectionOperator.PLAIN_SOCKET_FACTORY, + DefaultSchemePortResolver.INSTANCE, + dnsResolver, + (Lookup) name -> null, + true, + TimeValue.ofMilliseconds(20)); + + final DefaultManagedHttpClientConnection conn = new DefaultManagedHttpClientConnection( + "test", null, null, null, null, null, null, null, null); + final InetSocketAddress localAddress = new InetSocketAddress( + InetAddress.getLoopbackAddress(), explicitLocalPort); + + assertTimeoutPreemptively(Duration.ofSeconds(5), () -> + operator.connect(conn, new HttpHost("http", "somehost", 80), localAddress, + TimeValue.ofSeconds(5), SocketConfig.DEFAULT, new BasicHttpContext())); + + assertEquals(listeningPort, ((InetSocketAddress) conn.getRemoteAddress()).getPort()); + assertEquals(explicitLocalPort, ((InetSocketAddress) conn.getLocalAddress()).getPort()); + conn.close(); + } + } + + /** + * Regression test: on the Happy Eyeballs racing path, a TLS handshake failure after the + * TCP race has already picked a winner must still close that winning socket, exactly as + * the sequential path's own try/catch around bind + TLS upgrade already does. + */ + @Test + void happyEyeballsRacing_tlsHandshakeFailure_closesWinningSocket() throws Exception { + try (ServerSocket server = new ServerSocket(0, 1, InetAddress.getLoopbackAddress()); + ServerSocket deadPortProbe = new ServerSocket(0, 1, InetAddress.getLoopbackAddress())) { + final int listeningPort = server.getLocalPort(); + final int deadPort = deadPortProbe.getLocalPort(); + deadPortProbe.close(); + + final List remoteAddresses = Arrays.asList( + new InetSocketAddress(InetAddress.getLoopbackAddress(), deadPort), + new InetSocketAddress(InetAddress.getLoopbackAddress(), listeningPort)); + final DnsResolver dnsResolver = Mockito.mock(DnsResolver.class); + Mockito.when(dnsResolver.resolve(Mockito.anyString(), Mockito.anyInt())).thenReturn(remoteAddresses); + + final AtomicReference winningSocket = new AtomicReference<>(); + final DetachedSocketFactory trackingFactory = (final Proxy proxy) -> { + final Socket s = new Socket(); + winningSocket.set(s); + return s; + }; + + final TlsSocketStrategy failingTls = Mockito.mock(TlsSocketStrategy.class); + Mockito.when(failingTls.upgrade(Mockito.any(), Mockito.anyString(), Mockito.anyInt(), Mockito.any(), Mockito.any())) + .thenThrow(new IOException("handshake failed")); + final Lookup tlsLookup = scheme -> failingTls; + + final DefaultHttpClientConnectionOperator operator = new DefaultHttpClientConnectionOperator( + trackingFactory, DefaultSchemePortResolver.INSTANCE, dnsResolver, tlsLookup, true, TimeValue.ofMilliseconds(20)); + final DefaultManagedHttpClientConnection conn = new DefaultManagedHttpClientConnection( + "test", null, null, null, null, null, null, null, null); + + assertThrows(IOException.class, () -> + assertTimeoutPreemptively(Duration.ofSeconds(5), () -> + operator.connect(conn, new HttpHost("https", "somehost", 443), null, + TimeValue.ofSeconds(5), SocketConfig.DEFAULT, new BasicHttpContext()))); + + assertTrue(winningSocket.get().isClosed(), "the winning socket must be closed when the TLS handshake fails"); + } + } + + /** Same as above, but the TLS strategy throws a RuntimeException rather than an IOException. */ + @Test + void happyEyeballsRacing_tlsHandshakeRuntimeFailure_closesWinningSocket() throws Exception { + try (ServerSocket server = new ServerSocket(0, 1, InetAddress.getLoopbackAddress()); + ServerSocket deadPortProbe = new ServerSocket(0, 1, InetAddress.getLoopbackAddress())) { + final int listeningPort = server.getLocalPort(); + final int deadPort = deadPortProbe.getLocalPort(); + deadPortProbe.close(); + + final List remoteAddresses = Arrays.asList( + new InetSocketAddress(InetAddress.getLoopbackAddress(), deadPort), + new InetSocketAddress(InetAddress.getLoopbackAddress(), listeningPort)); + final DnsResolver dnsResolver = Mockito.mock(DnsResolver.class); + Mockito.when(dnsResolver.resolve(Mockito.anyString(), Mockito.anyInt())).thenReturn(remoteAddresses); + + final AtomicReference winningSocket = new AtomicReference<>(); + final DetachedSocketFactory trackingFactory = (final Proxy proxy) -> { + final Socket s = new Socket(); + winningSocket.set(s); + return s; + }; + + final TlsSocketStrategy failingTls = Mockito.mock(TlsSocketStrategy.class); + Mockito.when(failingTls.upgrade(Mockito.any(), Mockito.anyString(), Mockito.anyInt(), Mockito.any(), Mockito.any())) + .thenThrow(new IllegalStateException("handshake blew up")); + final Lookup tlsLookup = scheme -> failingTls; + + final DefaultHttpClientConnectionOperator operator = new DefaultHttpClientConnectionOperator( + trackingFactory, DefaultSchemePortResolver.INSTANCE, dnsResolver, tlsLookup, true, TimeValue.ofMilliseconds(20)); + final DefaultManagedHttpClientConnection conn = new DefaultManagedHttpClientConnection( + "test", null, null, null, null, null, null, null, null); + + assertThrows(IllegalStateException.class, () -> + assertTimeoutPreemptively(Duration.ofSeconds(5), () -> + operator.connect(conn, new HttpHost("https", "somehost", 443), null, + TimeValue.ofSeconds(5), SocketConfig.DEFAULT, new BasicHttpContext()))); + + assertTrue(winningSocket.get().isClosed(), "the winning socket must be closed when the TLS handshake throws unexpectedly"); + } + } +} diff --git a/httpclient5/src/test/java/org/apache/hc/client5/http/impl/io/HappyEyeballsSocketConnectorTest.java b/httpclient5/src/test/java/org/apache/hc/client5/http/impl/io/HappyEyeballsSocketConnectorTest.java new file mode 100644 index 0000000000..eec39052ec --- /dev/null +++ b/httpclient5/src/test/java/org/apache/hc/client5/http/impl/io/HappyEyeballsSocketConnectorTest.java @@ -0,0 +1,340 @@ +/* + * ==================================================================== + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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. + * ==================================================================== + * + * This software consists of voluntary contributions made by many + * individuals on behalf of the Apache Software Foundation. For more + * information on the Apache Software Foundation, please see + * . + * + */ +package org.apache.hc.client5.http.impl.io; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.IOException; +import java.io.InterruptedIOException; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.net.SocketAddress; +import java.net.SocketException; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +import org.apache.hc.core5.http.HttpHost; +import org.apache.hc.core5.util.TimeValue; +import org.junit.jupiter.api.Test; + +class HappyEyeballsSocketConnectorTest { + + /** A socket whose {@link #connect} blocks until {@link #close} is called, then fails, matching the real behavior of closing a socket mid-connect. */ + private static final class BlockingSocket extends Socket { + + private final CountDownLatch closed = new CountDownLatch(1); + volatile boolean connectAttempted; + + @Override + public void connect(final SocketAddress endpoint, final int timeout) throws IOException { + connectAttempted = true; + try { + closed.await(); + } catch (final InterruptedException ex) { + Thread.currentThread().interrupt(); + } + throw new SocketException("Socket closed"); + } + + @Override + public synchronized void close() throws IOException { + closed.countDown(); + super.close(); + } + } + + 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; + } + + @Test + void singleAddress_connectsWithoutRacing() throws Exception { + final HappyEyeballsSocketConnector connector = new HappyEyeballsSocketConnector(TimeValue.ofMilliseconds(20)); + final Socket expected = new Socket(); + final AtomicInteger attempts = new AtomicInteger(); + + final Socket actual = connector.connect(new HttpHost("somehost"), addresses(1), (remoteAddress, socketConsumer) -> { + attempts.incrementAndGet(); + socketConsumer.accept(expected); + return expected; + }); + + assertEquals(expected, actual); + assertEquals(1, attempts.get()); + } + + @Test + void secondAddressWinsWhileFirstStalls_loserSocketIsClosed() throws Exception { + final HappyEyeballsSocketConnector connector = new HappyEyeballsSocketConnector(TimeValue.ofMilliseconds(20)); + final BlockingSocket firstSocket = new BlockingSocket(); + final Socket secondSocket = new Socket(); + + final Socket winner = assertTimeoutPreemptively(Duration.ofSeconds(5), () -> + connector.connect(new HttpHost("somehost"), addresses(2), (remoteAddress, socketConsumer) -> { + if (remoteAddress.getAddress().getAddress()[3] == 1) { + socketConsumer.accept(firstSocket); + firstSocket.connect(remoteAddress, 0); // blocks until closed by the race, then throws + return firstSocket; + } + socketConsumer.accept(secondSocket); + return secondSocket; + })); + + assertEquals(secondSocket, winner); + assertTrue(firstSocket.connectAttempted, "the first attempt must have genuinely started connecting"); + assertTrue(firstSocket.isClosed(), "the losing, still-connecting socket must be closed rather than leaked"); + } + + @Test + void allAddressesFail_aggregatedFailure() throws Exception { + final HappyEyeballsSocketConnector connector = new HappyEyeballsSocketConnector(TimeValue.ofMilliseconds(20)); + + final IOException ex = assertThrows(IOException.class, () -> + connector.connect(new HttpHost("somehost"), addresses(2), (remoteAddress, socketConsumer) -> { + throw new IOException("refused: " + remoteAddress); + })); + assertTrue(ex.getMessage() != null); + } + + @Test + void firstAddressFailsFast_secondStartsWithoutWaitingOutTheFullDelay() throws Exception { + final HappyEyeballsSocketConnector connector = new HappyEyeballsSocketConnector(TimeValue.ofMilliseconds(250)); + final Socket winnerSocket = new Socket(); + + final long start = System.nanoTime(); + final Socket winner = assertTimeoutPreemptively(Duration.ofSeconds(5), () -> + connector.connect(new HttpHost("somehost"), addresses(2), (remoteAddress, socketConsumer) -> { + if (remoteAddress.getAddress().getAddress()[3] == 1) { + throw new IOException("refused"); + } + socketConsumer.accept(winnerSocket); + return winnerSocket; + })); + final long elapsedMs = (System.nanoTime() - start) / 1_000_000; + + // The 2nd address must not have had to wait out the 250ms stagger delay, since the + // 1st address failed fast rather than merely being slow: it only waits the RFC 8305 + // minimum inter-attempt spacing (10ms). + assertEquals(winnerSocket, winner); + assertTrue(elapsedMs < 200, "must advance well before the full stagger delay elapses, took " + elapsedMs + "ms"); + } + + @Test + void interruptionWhileStillRacing_closesLoserSocket() throws Exception { + final HappyEyeballsSocketConnector connector = new HappyEyeballsSocketConnector(TimeValue.ofMilliseconds(20)); + final BlockingSocket firstSocket = new BlockingSocket(); + final Socket secondSocket = new Socket(); + + final Socket winner = assertTimeoutPreemptively(Duration.ofSeconds(5), () -> + connector.connect(new HttpHost("somehost"), addresses(2), (remoteAddress, socketConsumer) -> { + if (remoteAddress.getAddress().getAddress()[3] == 1) { + socketConsumer.accept(firstSocket); + firstSocket.connect(remoteAddress, 0); + return firstSocket; + } + socketConsumer.accept(secondSocket); + return secondSocket; + })); + + assertEquals(secondSocket, winner); + // The executor's shutdownNow() (called in the finally block) interrupts the still + // "connecting" worker; its socket must still end up closed, not leaked. + assertTrue(firstSocket.isClosed()); + } + + @Test + void interruptionDuringMinimumSpacingWait_abortsBeforeStartingNextAddress() throws Exception { + // Regression test: sleepOrAbort() (the minimum-spacing wait after a fast failure) must + // propagate an interruption instead of swallowing it and continuing the race. + final HappyEyeballsSocketConnector connector = new HappyEyeballsSocketConnector(TimeValue.ofMilliseconds(250)); + final AtomicInteger attempts = new AtomicInteger(); + final CountDownLatch firstFailed = new CountDownLatch(1); + final AtomicReference thrown = new AtomicReference<>(); + + final Thread caller = new Thread(() -> { + try { + connector.connect(new HttpHost("somehost"), addresses(2), (remoteAddress, socketConsumer) -> { + attempts.incrementAndGet(); + if (remoteAddress.getAddress().getAddress()[3] == 1) { + firstFailed.countDown(); + throw new IOException("refused"); + } + return new Socket(); + }); + } catch (final Throwable ex) { + thrown.set(ex); + } + }); + caller.start(); + assertTrue(firstFailed.await(5, TimeUnit.SECONDS)); + caller.interrupt(); // races the <=10ms minimum-spacing wait that follows the failure + caller.join(5000); + + assertFalse(caller.isAlive()); + // Best-effort timing: if the interruption landed while connect() was still running, it + // must be reported as InterruptedIOException, never silently swallowed. + if (thrown.get() != null) { + assertTrue(thrown.get() instanceof InterruptedIOException, + "an interruption during racing must surface as InterruptedIOException: " + thrown.get()); + } + } + + @Test + void interruptingTheCallingThread_abortsAndClosesTheStillConnectingSocket() throws Exception { + final HappyEyeballsSocketConnector connector = new HappyEyeballsSocketConnector(TimeValue.ofMilliseconds(20)); + final BlockingSocket onlySocket = new BlockingSocket(); + final CountDownLatch connecting = new CountDownLatch(1); + final AtomicReference thrown = new AtomicReference<>(); + + final Thread caller = new Thread(() -> { + try { + connector.connect(new HttpHost("somehost"), addresses(1), (remoteAddress, socketConsumer) -> { + socketConsumer.accept(onlySocket); + connecting.countDown(); + onlySocket.connect(remoteAddress, 0); // blocks until closed or interrupted + return onlySocket; + }); + } catch (final Throwable ex) { + thrown.set(ex); + } + }); + caller.start(); + assertTrue(connecting.await(5, TimeUnit.SECONDS), "the attempt must have genuinely started connecting"); + caller.interrupt(); + caller.join(5000); + + assertFalse(caller.isAlive(), "the calling thread must have returned after being interrupted"); + assertTrue(thrown.get() instanceof InterruptedIOException, + "connect() must report the interruption rather than silently returning: was " + thrown.get()); + assertTrue(onlySocket.isClosed(), "the still-connecting socket must be closed, not leaked, on interruption"); + } + + @Test + void registrationAfterRaceDecided_closesSocketImmediately() throws Exception { + final HappyEyeballsSocketConnector connector = new HappyEyeballsSocketConnector(TimeValue.ofMilliseconds(20)); + final Socket winnerSocket = new Socket(); + final CountDownLatch registerNow = new CountDownLatch(1); + final CountDownLatch registered = new CountDownLatch(1); + final AtomicReference lateSocketRef = new AtomicReference<>(); + + final Socket winner = assertTimeoutPreemptively(Duration.ofSeconds(5), () -> + connector.connect(new HttpHost("somehost"), addresses(2), (remoteAddress, socketConsumer) -> { + if (remoteAddress.getAddress().getAddress()[3] == 1) { + // Registers strictly after the race below has already picked a winner. + awaitUninterruptibly(registerNow); + final Socket lateSocket = new Socket(); + lateSocketRef.set(lateSocket); + socketConsumer.accept(lateSocket); + registered.countDown(); + throw new IOException("too late"); + } + socketConsumer.accept(winnerSocket); + return winnerSocket; + })); + + assertEquals(winnerSocket, winner); + registerNow.countDown(); + assertTrue(registered.await(5, TimeUnit.SECONDS)); + assertTrue(lateSocketRef.get().isClosed(), + "a socket registered strictly after the race is decided must be closed immediately, not left open"); + } + + private static void awaitUninterruptibly(final CountDownLatch latch) { + try { + latch.await(); + } catch (final InterruptedException ex) { + Thread.currentThread().interrupt(); + } + } + + @Test + void doesNotRejectSingleAddressBelowStaggerIndex() throws Exception { + final HappyEyeballsSocketConnector connector = new HappyEyeballsSocketConnector(TimeValue.ofMilliseconds(20)); + final Socket socket = new Socket(); + + final Socket result = connector.connect(new HttpHost("somehost"), addresses(1), (remoteAddress, socketConsumer) -> { + socketConsumer.accept(socket); + return socket; + }); + + assertEquals(socket, result); + assertFalse(socket.isClosed()); + } + + /** + * Real-socket integration test (no mocks, no fakes): one address has nothing listening + * (genuine, OS-level immediate connection refusal) and the other is a real local server + * that accepts. Proves actual fallover over real loopback TCP, not just that the first + * address happens to succeed. + */ + @Test + void realSockets_unreachableAddressFailsOver_toRealListeningServer() throws Exception { + try (ServerSocket server = new ServerSocket(0, 1, InetAddress.getLoopbackAddress())) { + final int listeningPort = server.getLocalPort(); + final int deadPort = findUnusedLoopbackPort(); + + final HappyEyeballsSocketConnector connector = new HappyEyeballsSocketConnector(TimeValue.ofMilliseconds(20)); + final List targets = new ArrayList<>(); + targets.add(new InetSocketAddress(InetAddress.getLoopbackAddress(), deadPort)); + targets.add(new InetSocketAddress(InetAddress.getLoopbackAddress(), listeningPort)); + + final Socket winner = assertTimeoutPreemptively(Duration.ofSeconds(5), () -> + connector.connect(new HttpHost("localhost"), targets, (remoteAddress, socketConsumer) -> { + final Socket s = new Socket(); + socketConsumer.accept(s); + s.connect(remoteAddress, 2000); + return s; + })); + + assertEquals(listeningPort, winner.getPort()); + assertFalse(winner.isClosed()); + winner.close(); + } + } + + /** Finds a loopback TCP port nothing is listening on, by briefly binding then closing it. */ + private static int findUnusedLoopbackPort() throws IOException { + try (ServerSocket probe = new ServerSocket(0, 1, InetAddress.getLoopbackAddress())) { + return probe.getLocalPort(); + } + } +}