JAVA-6194 Add MongoSocksProxyException for CMAP backpressure labeling

driver-core/src/main/com/mongodb/MongoSocksProxyException.java

@@ -0,0 +1,178 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * Licensed 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.
+ */
+
+package com.mongodb;
+
+import com.mongodb.lang.Nullable;
+
+import static com.mongodb.assertions.Assertions.notNull;
+
+/**
+ * Thrown when an error occurs while establishing a connection to a SOCKS5 proxy.
+ *
+ * <p>The {@link #getHandshakePhase()} identifies which phase of the SOCKS5 handshake failed.
+ * {@link #getProxyReplyCode()} returns the RFC 1928 reply code sent by the proxy when a
+ * non-success CONNECT reply was successfully parsed; it returns {@code null} otherwise
+ * (including for {@link HandshakePhase#CONNECT_RELAY} failures caused by an I/O error or
+ * an unrecognized reply field).
+ *
+ * <p>RFC 1928 reply codes: 1=general failure, 2=connection not allowed by ruleset,
+ * 3=network unreachable, 4=host unreachable, 5=connection refused, 6=TTL expired,
+ * 7=command not supported, 8=address type not supported.
+ *
+ * <p>Constructor parameter ordering follows the parent class first (message, address,
+ * optional cause), then SOCKS-specific arguments (handshakePhase, optional proxyReplyCode).
+ *
+ * @since 5.8
+ */
+public class MongoSocksProxyException extends MongoSocketOpenException {
+    private static final long serialVersionUID = 1L;
+
+    /**
+     * The phase of the SOCKS5 handshake at which the failure occurred.
+     *
+     * @since 5.8
+     */
+    public enum HandshakePhase {
+        /**
+         * TCP connection to the proxy host itself failed before any SOCKS5 exchange.
+         * The proxy may be temporarily unreachable.
+         */
+        PROXY_TCP_CONNECT,
+
+        /**
+         * The SOCKS5 method-selection exchange failed. Causes include: incompatible
+         * proxy version, no common authentication method, an unrecognized method, or
+         * an I/O failure (EOF, timeout, broken pipe) while sending the method-selection
+         * request or reading its reply.
+         */
+        NEGOTIATION,
+
+        /**
+         * Username/password sub-negotiation with the proxy failed. Causes include:
+         * the proxy rejecting the credentials (typically wrong username/password),
+         * or an I/O failure (EOF, timeout, broken pipe) while sending credentials
+         * or reading the auth result.
+         */
+        AUTHENTICATION,
+
+        /**
+         * A failure occurred while sending the CONNECT request to the proxy or
+         * reading/parsing its reply. Causes include: a parsed non-success RFC 1928
+         * reply (in which case {@link MongoSocksProxyException#getProxyReplyCode()}
+         * carries the code), an unrecognized reply field or address type, or an
+         * I/O failure (EOF, timeout, broken pipe) on the CONNECT exchange.
+         */
+        CONNECT_RELAY
+    }
+
+    private final HandshakePhase handshakePhase;
+
+    @Nullable
+    private final Integer proxyReplyCode;
+
+    /**
+     * Construct an instance with no RFC 1928 reply code and no cause. Suitable for any phase
+     * whose failure does not carry a parsed reply code: {@link HandshakePhase#PROXY_TCP_CONNECT},
+     * {@link HandshakePhase#NEGOTIATION}, {@link HandshakePhase#AUTHENTICATION}, and the
+     * {@link HandshakePhase#CONNECT_RELAY} sub-cases driven by an I/O failure or an unrecognized
+     * reply field.
+     *
+     * @param message        the message
+     * @param serverAddress  the server address
+     * @param handshakePhase the phase at which the failure occurred
+     */
+    public MongoSocksProxyException(final String message, final ServerAddress serverAddress, final HandshakePhase handshakePhase) {
+        this(message, serverAddress, notNull("handshakePhase", handshakePhase), null);
+    }
+
+    /**
+     * Construct an instance with no RFC 1928 reply code. Suitable for any phase whose failure
+     * does not carry a parsed reply code: {@link HandshakePhase#PROXY_TCP_CONNECT},
+     * {@link HandshakePhase#NEGOTIATION}, {@link HandshakePhase#AUTHENTICATION}, and the
+     * {@link HandshakePhase#CONNECT_RELAY} sub-cases driven by an I/O failure or an unrecognized
+     * reply field.
+     *
+     * @param message        the message
+     * @param address        the server address
+     * @param cause          the cause
+     * @param handshakePhase the phase at which the failure occurred
+     */
+    public MongoSocksProxyException(final String message, final ServerAddress address,
+                                    final Throwable cause, final HandshakePhase handshakePhase) {
+        this(message, address, cause, notNull("handshakePhase", handshakePhase), null);
+    }
+
+    /**
+     * Construct an instance with an optional RFC 1928 reply code. A non-{@code null}
+     * {@code proxyReplyCode} should only accompany {@link HandshakePhase#CONNECT_RELAY} and
+     * indicates a successfully parsed non-success reply from the proxy. Use {@code null} in
+     * all other cases — including {@link HandshakePhase#CONNECT_RELAY} failures caused by an
+     * I/O error or an unrecognized reply field.
+     *
+     * @param message        the message
+     * @param address        the server address
+     * @param handshakePhase the phase at which the failure occurred
+     * @param proxyReplyCode the RFC 1928 reply code, or {@code null}
+     */
+    public MongoSocksProxyException(final String message, final ServerAddress address, final HandshakePhase handshakePhase,
+            @Nullable final Integer proxyReplyCode) {
+        super(message, address);
+        this.handshakePhase = notNull("handshakePhase", handshakePhase);
+        this.proxyReplyCode = proxyReplyCode;
+    }
+
+    /**
+     * Construct an instance with an optional RFC 1928 reply code. A non-{@code null}
+     * {@code proxyReplyCode} should only accompany {@link HandshakePhase#CONNECT_RELAY} and
+     * indicates a successfully parsed non-success reply from the proxy. Use {@code null} in
+     * all other cases — including {@link HandshakePhase#CONNECT_RELAY} failures caused by an
+     * I/O error or an unrecognized reply field.
+     *
+     * @param message        the message
+     * @param address        the server address
+     * @param cause          the cause
+     * @param handshakePhase the phase at which the failure occurred
+     * @param proxyReplyCode the RFC 1928 reply code, or {@code null}
+     */
+    public MongoSocksProxyException(final String message, final ServerAddress address,
+                                    final Throwable cause, final HandshakePhase handshakePhase,
+                                    @Nullable final Integer proxyReplyCode) {
+        super(message, address, cause);
+        this.handshakePhase = notNull("handshakePhase", handshakePhase);
+        this.proxyReplyCode = proxyReplyCode;
+    }
+
+    /**
+     * Returns the phase of the SOCKS5 handshake at which the failure occurred.
+     *
+     * @return the handshake phase, never {@code null}
+     */
+    public HandshakePhase getHandshakePhase() {
+        return handshakePhase;
+    }
+
+    /**
+     * Returns the RFC 1928 reply code sent by the SOCKS5 proxy when a non-success CONNECT
+     * reply was successfully parsed, or {@code null} otherwise.
+     *
+     * @return the RFC 1928 proxy reply code, or {@code null}
+     */
+    @Nullable
+    public Integer getProxyReplyCode() {
+        return proxyReplyCode;
+    }
+}

driver-core/src/main/com/mongodb/internal/connection/BackpressureErrorLabeler.java

@@ -18,6 +18,7 @@
 
 import com.mongodb.MongoException;
 import com.mongodb.MongoSocketException;
+import com.mongodb.MongoSocksProxyException;
 
 import javax.net.ssl.SSLHandshakeException;
 import javax.net.ssl.SSLPeerUnverifiedException;
@@ -76,19 +77,36 @@ static void applyLabelsIfEligible(final Throwable t) {
             return;
         }
         MongoSocketException socketException = (MongoSocketException) t;
+        if (isSocksFailure(socketException)) {
+            return;
+        }
         if (isDnsLookupFailure(socketException)) {
             return;
         }
         if (isTlsConfigurationError(socketException)) {
             return;
         }
-        // TODO-BACKPRESSURE Nabil - Add SOCKS5 check once JAVA-6194 is introduced
-        // async proxy error surfaces can be handled together — likely via a dedicated internal
-        // exception thrown from the proxy code path.
         socketException.addLabel(MongoException.SYSTEM_OVERLOADED_ERROR_LABEL);
         socketException.addLabel(MongoException.RETRYABLE_ERROR_LABEL);
     }
 
+    private static boolean isSocksFailure(final MongoSocketException t) {
+        if (!(t instanceof MongoSocksProxyException)) {
+            return false;
+        }
+        MongoSocksProxyException socksException = (MongoSocksProxyException) t;
+        if (socksException.getHandshakePhase() != MongoSocksProxyException.HandshakePhase.CONNECT_RELAY) {
+            return true;
+        }
+        Integer replyCode = socksException.getProxyReplyCode();
+        if (replyCode == null) {
+            return true;
+        }
+        return replyCode != SocksSocket.ServerReply.NET_UNREACHABLE.getReplyNumber()
+                && replyCode != SocksSocket.ServerReply.HOST_UNREACHABLE.getReplyNumber()
+                && replyCode != SocksSocket.ServerReply.CONN_REFUSED.getReplyNumber();
+    }
+
     private static boolean isDnsLookupFailure(final MongoSocketException t) {
         Throwable cause = t.getCause();
         while (cause != null) {

driver-core/src/main/com/mongodb/internal/connection/SocketStream.java

@@ -16,9 +16,11 @@
 
 package com.mongodb.internal.connection;
 
+import com.mongodb.MongoInterruptedException;
 import com.mongodb.MongoSocketException;
 import com.mongodb.MongoSocketOpenException;
 import com.mongodb.MongoSocketReadException;
+import com.mongodb.MongoSocksProxyException;
 import com.mongodb.ServerAddress;
 import com.mongodb.connection.AsyncCompletionHandler;
 import com.mongodb.connection.ProxySettings;
@@ -38,6 +40,7 @@
 import java.net.SocketTimeoutException;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Optional;
 
 import static com.mongodb.assertions.Assertions.assertTrue;
 import static com.mongodb.assertions.Assertions.notNull;
@@ -79,10 +82,16 @@ public void open(final OperationContext operationContext) {
             socket = initializeSocket(operationContext);
             outputStream = socket.getOutputStream();
             inputStream = socket.getInputStream();
+        } catch (MongoSocksProxyException e) {
+            close();
+            throw e;
         } catch (IOException e) {
             close();
-            throw translateInterruptedException(e, "Interrupted while connecting")
-                    .orElseThrow(() -> new MongoSocketOpenException("Exception opening socket", getAddress(), e));
+            Optional<MongoInterruptedException> interrupted = translateInterruptedException(e, "Interrupted while connecting");
+            if (interrupted.isPresent()) {
+                throw interrupted.get();
+            }
+            throw new MongoSocketOpenException("Exception opening socket", getAddress(), e);
         }
     }
 
@@ -119,15 +128,32 @@ private SSLSocket initializeSslSocketOverSocksProxy(final OperationContext opera
         final int serverPort = address.getPort();
 
         SocksSocket socksProxy = new SocksSocket(settings.getProxySettings());
-        configureSocket(socksProxy, operationContext, settings);
-        InetSocketAddress inetSocketAddress = toSocketAddress(serverHost, serverPort);
-        socksProxy.connect(inetSocketAddress, operationContext.getTimeoutContext().getConnectTimeoutMs());
-
-        SSLSocket sslSocket = (SSLSocket) sslSocketFactory.createSocket(socksProxy, serverHost, serverPort, true);
-        //Even though Socks proxy connection is already established, TLS handshake has not been performed yet.
-        //So it is possible to set SSL parameters before handshake is done.
-        configureSslSocket(sslSocket, sslSettings, inetSocketAddress);
-        return sslSocket;
+        // Track the outermost socket layer to close on failure. Initially this is socksProxy;
+        // once we wrap it into an SSLSocket, that becomes the outermost layer and closing it
+        // tears down the underlying socksProxy as well.
+        Socket toClose = socksProxy;
+        try {
+            configureSocket(socksProxy, operationContext, settings);
+            InetSocketAddress inetSocketAddress = toSocketAddress(serverHost, serverPort);
+            try {
+                socksProxy.connect(inetSocketAddress, operationContext.getTimeoutContext().getConnectTimeoutMs());
+            } catch (IOException e) {
+                throw wrapAsProxyTcpConnect(e);
+            }
+            SSLSocket sslSocket = (SSLSocket) sslSocketFactory.createSocket(socksProxy, serverHost, serverPort, true);
+            toClose = sslSocket;
+            //Even though Socks proxy connection is already established, TLS handshake has not been performed yet.
+            //So it is possible to set SSL parameters before handshake is done.
+            configureSslSocket(sslSocket, sslSettings, inetSocketAddress);
+            return sslSocket;
+        } catch (IOException | RuntimeException e) {
+            try {
+                toClose.close();
+            } catch (IOException closeException) {
+                e.addSuppressed(closeException);
+            }
+            throw e;
+        }
     }
 
 
@@ -139,19 +165,41 @@ private static InetSocketAddress toSocketAddress(final String serverHost, final
         return InetSocketAddress.createUnresolved(serverHost, serverPort);
     }
 
+    private MongoSocksProxyException wrapAsProxyTcpConnect(final IOException cause) {
+        ProxySettings proxySettings = settings.getProxySettings();
+        return new MongoSocksProxyException(
+                "Exception connecting to SOCKS5 proxy (" + proxySettings.getHost() + ":" + proxySettings.getPort() + ")",
+                getAddress(), cause,
+                MongoSocksProxyException.HandshakePhase.PROXY_TCP_CONNECT);
+    }
+
     private Socket initializeSocketOverSocksProxy(final OperationContext operationContext) throws IOException {
         Socket createdSocket = socketFactory.createSocket();
-        configureSocket(createdSocket, operationContext, settings);
-        /*
-          Wrap the configured socket with SocksSocket to add extra functionality.
-          Reason for separate steps: We can't directly extend Java 11 methods within 'SocksSocket'
-          to configure itself.
-         */
-        SocksSocket socksProxy = new SocksSocket(createdSocket, settings.getProxySettings());
-
-        socksProxy.connect(toSocketAddress(address.getHost(), address.getPort()),
-                operationContext.getTimeoutContext().getConnectTimeoutMs());
-        return socksProxy;
+        try {
+            configureSocket(createdSocket, operationContext, settings);
+            /*
+              Wrap the configured socket with SocksSocket to add extra functionality.
+              Reason for separate steps: We can't directly extend Java 11 methods within 'SocksSocket'
+              to configure itself.
+             */
+            SocksSocket socksProxy = new SocksSocket(createdSocket, settings.getProxySettings());
+            try {
+                socksProxy.connect(toSocketAddress(address.getHost(), address.getPort()),
+                        operationContext.getTimeoutContext().getConnectTimeoutMs());
+            } catch (IOException e) {
+                throw wrapAsProxyTcpConnect(e);
+            }
+            return socksProxy;
+        } catch (IOException | RuntimeException e) {
+            // SocksSocket.connect() closes itself on failure, but createdSocket may not yet
+            // be owned by a SocksSocket (e.g. configureSocket threw). Close defensively;
+            try {
+                createdSocket.close();
+            } catch (IOException closeException) {
+                e.addSuppressed(closeException);
+            }
+            throw e;
+        }
     }
 
     @Override