JAVA-5949 preserve connection pool on backpressure errors when establishing connections

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

@@ -0,0 +1,102 @@
+/*
+ * 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.internal.connection;
+
+import com.mongodb.MongoException;
+import com.mongodb.MongoSocketException;
+
+import javax.net.ssl.SSLHandshakeException;
+import javax.net.ssl.SSLPeerUnverifiedException;
+import javax.net.ssl.SSLProtocolException;
+import java.net.UnknownHostException;
+import java.security.cert.CertPathBuilderException;
+import java.security.cert.CertPathValidatorException;
+import java.security.cert.CertificateException;
+import java.util.Locale;
+
+/**
+ * Attaches {@link MongoException#SYSTEM_OVERLOADED_ERROR_LABEL} and
+ * {@link MongoException#RETRYABLE_ERROR_LABEL} to network errors encountered during connection
+ * establishment or the hello message, per the CMAP specification.
+ *
+ * <p>This is topology-agnostic: it must be invoked from the connection-establishment path so that
+ * both default SDAM and load-balanced modes are covered.
+ */
+final class BackpressureErrorLabeler {
+
+    private BackpressureErrorLabeler() {
+    }
+
+    static void applyLabelsIfEligible(final Throwable t) {
+        if (!(t instanceof MongoSocketException)) {
+            return;
+        }
+        if (isDnsLookupFailure(t)) {
+            return;
+        }
+        if (isTlsConfigurationError(t)) {
+            return;
+        }
+        // TODO-BACKPRESSURE Nabil - SOCKS5 Revisit alongside JAVA-5205 (SOCKS5 in async) so both sync and
+        // async proxy error surfaces can be handled together — likely via a dedicated internal
+        // exception thrown from the proxy code path.
+        MongoException mongoException = (MongoException) t;
+        mongoException.addLabel(MongoException.SYSTEM_OVERLOADED_ERROR_LABEL);
+        mongoException.addLabel(MongoException.RETRYABLE_ERROR_LABEL);
+    }
+
+    private static boolean isDnsLookupFailure(final Throwable t) {
+        Throwable cause = t.getCause();
+        while (cause != null) {
+            if (cause instanceof UnknownHostException) {
+                return true;
+            }
+            cause = cause.getCause();
+        }
+        return false;
+    }
+
+    static boolean isTlsConfigurationError(final Throwable t) {
+        Throwable cause = t.getCause();
+        while (cause != null) {
+            if (cause instanceof CertificateException
+                    || cause instanceof CertPathBuilderException
+                    || cause instanceof CertPathValidatorException
+                    || cause instanceof SSLPeerUnverifiedException
+                    || cause instanceof SSLProtocolException) {
+                return true;
+            }
+            if (cause instanceof SSLHandshakeException) {
+                String message = cause.getMessage();
+                if (message != null) {
+                    String lowerMessage = message.toLowerCase(Locale.ROOT);
+                    if (lowerMessage.contains("certificate")
+                            || lowerMessage.contains("verify")
+                            || lowerMessage.contains("trust")
+                            || lowerMessage.contains("hostname")
+                            || lowerMessage.contains("protocol")
+                            || lowerMessage.contains("cipher")
+                            || lowerMessage.contains("handshake_failure")) {
+                        return true;
+                    }
+                }
+            }
+            cause = cause.getCause();
+        }
+        return false;
+    }
+}

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

@@ -137,6 +137,9 @@ private void handleException(final SdamIssue sdamIssue, final boolean beforeHand
             serverMonitor.connect();
         } else if (sdamIssue.relatedToNetworkNotTimeout()
                 || (beforeHandshake && (sdamIssue.relatedToNetworkTimeout() || sdamIssue.relatedToAuth()))) {
+            if (sdamIssue.hasBackpressureLabel()) {
+                return;
+            }
             updateDescription(sdamIssue.serverDescription());
             connectionPool.invalidate(sdamIssue.exception().orElse(null));
             serverMonitor.cancelCurrentCheck();

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

@@ -230,9 +230,11 @@ public void open(final OperationContext originalOperationContext) {
         isTrue("Open already called", stream == null);
         stream = streamFactory.create(serverId.getAddress());
         OperationContext operationContext = originalOperationContext;
+        boolean beforeHandshake = true;
         try {
             stream.open(operationContext);
             InternalConnectionInitializationDescription initializationDescription = connectionInitializer.startHandshake(this, operationContext);
+            beforeHandshake = false;
 
             operationContext = operationContext.withOverride(TimeoutContext::withNewlyStartedMaintenanceTimeout);
             initAfterHandshakeStart(initializationDescription);
@@ -241,6 +243,9 @@ public void open(final OperationContext originalOperationContext) {
             initAfterHandshakeFinish(initializationDescription);
         } catch (Throwable t) {
             close();
+            if (beforeHandshake) {
+                BackpressureErrorLabeler.applyLabelsIfEligible(t);
+            }
             if (t instanceof MongoException) {
                 throw (MongoException) t;
             } else {
@@ -263,6 +268,7 @@ public void completed(@Nullable final Void aVoid) {
                             (initialResult, initialException) -> {
                                     if (initialException != null) {
                                         close();
+                                        BackpressureErrorLabeler.applyLabelsIfEligible(initialException);
                                         callback.onResult(null, initialException);
                                     } else {
                                         assertNotNull(initialResult);
@@ -278,11 +284,13 @@ public void completed(@Nullable final Void aVoid) {
                 @Override
                 public void failed(final Throwable t) {
                     close();
+                    BackpressureErrorLabeler.applyLabelsIfEligible(t);
                     callback.onResult(null, t);
                 }
             });
         } catch (Throwable t) {
             close();
+            BackpressureErrorLabeler.applyLabelsIfEligible(t);
             callback.onResult(null, t);
         }
     }

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

@@ -17,6 +17,7 @@
 package com.mongodb.internal.connection;
 
 import com.mongodb.MongoCommandException;
+import com.mongodb.MongoException;
 import com.mongodb.MongoNodeIsRecoveringException;
 import com.mongodb.MongoNotPrimaryException;
 import com.mongodb.MongoSecurityException;
@@ -162,6 +163,11 @@ boolean relatedToWriteConcern() {
             return exception instanceof MongoWriteConcernWithResponseException;
         }
 
+        boolean hasBackpressureLabel() {
+            return exception instanceof MongoException
+                    && ((MongoException) exception).hasErrorLabel(MongoException.SYSTEM_OVERLOADED_ERROR_LABEL);
+        }
+
         private static boolean stale(@Nullable final Throwable t, final ServerDescription currentServerDescription) {
             return TopologyVersionHelper.topologyVersion(t)
                     .map(candidateTopologyVersion -> TopologyVersionHelper.newerOrEqual(

driver-core/src/test/unit/com/mongodb/internal/connection/AbstractServerDiscoveryAndMonitoringTest.java

@@ -113,6 +113,7 @@ protected void applyApplicationError(final BsonDocument applicationError) {
 
         switch (when) {
             case "beforeHandshakeCompletes":
+                BackpressureErrorLabeler.applyLabelsIfEligible(exception);
                 server.sdamServerDescriptionManager().handleExceptionBeforeHandshake(
                         SdamIssue.of(exception, new SdamIssue.Context(server.serverId(), errorGeneration, maxWireVersion)));
                 break;

driver-core/src/test/unit/com/mongodb/internal/connection/BackpressureErrorLabelerTest.java

@@ -0,0 +1,156 @@
+/*
+ * 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.internal.connection;
+
+import com.mongodb.MongoCredential;
+import com.mongodb.MongoException;
+import com.mongodb.MongoSecurityException;
+import com.mongodb.MongoSocketException;
+import com.mongodb.MongoSocketOpenException;
+import com.mongodb.MongoSocketReadTimeoutException;
+import com.mongodb.ServerAddress;
+import org.bson.BsonDocument;
+import org.junit.jupiter.api.Test;
+
+import javax.net.ssl.SSLHandshakeException;
+import javax.net.ssl.SSLPeerUnverifiedException;
+import java.io.IOException;
+import java.net.UnknownHostException;
+import java.security.cert.CertificateException;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class BackpressureErrorLabelerTest {
+
+    private static final ServerAddress ADDRESS = new ServerAddress();
+
+    @Test
+    void mongoSocketExceptionIsLabeled() {
+        MongoSocketException e = new MongoSocketException("boom", ADDRESS);
+        BackpressureErrorLabeler.applyLabelsIfEligible(e);
+        assertHasBackpressureLabels(e);
+    }
+
+    @Test
+    void mongoSocketReadTimeoutIsLabeled() {
+        MongoSocketReadTimeoutException e = new MongoSocketReadTimeoutException("slow", ADDRESS, new IOException("read timed out"));
+        BackpressureErrorLabeler.applyLabelsIfEligible(e);
+        assertHasBackpressureLabels(e);
+    }
+
+    @Test
+    void mongoSocketOpenExceptionIsLabeled() {
+        MongoSocketOpenException e = new MongoSocketOpenException("open failed", ADDRESS, new IOException("connection refused"));
+        BackpressureErrorLabeler.applyLabelsIfEligible(e);
+        assertHasBackpressureLabels(e);
+    }
+
+    @Test
+    void dnsFailureIsNotLabeled() {
+        MongoSocketException e = new MongoSocketException("lookup failed", ADDRESS, new UnknownHostException("nope"));
+        BackpressureErrorLabeler.applyLabelsIfEligible(e);
+        assertLacksBackpressureLabels(e);
+    }
+
+    @Test
+    void dnsFailureDeepInCauseChainIsNotLabeled() {
+        Throwable dns = new UnknownHostException("nope");
+        IOException wrap1 = new IOException("wrap1", dns);
+        MongoSocketException e = new MongoSocketException("wrap2", ADDRESS, wrap1);
+        BackpressureErrorLabeler.applyLabelsIfEligible(e);
+        assertLacksBackpressureLabels(e);
+    }
+
+    @Test
+    void tlsCertificateErrorIsNotLabeled() {
+        CertificateException cert = new CertificateException("bad cert");
+        MongoSocketException e = new MongoSocketException("tls", ADDRESS, cert);
+        BackpressureErrorLabeler.applyLabelsIfEligible(e);
+        assertLacksBackpressureLabels(e);
+    }
+
+    @Test
+    void tlsPeerUnverifiedIsNotLabeled() {
+        SSLPeerUnverifiedException tls = new SSLPeerUnverifiedException("peer not verified");
+        MongoSocketException e = new MongoSocketException("tls", ADDRESS, tls);
+        BackpressureErrorLabeler.applyLabelsIfEligible(e);
+        assertLacksBackpressureLabels(e);
+    }
+
+    @Test
+    void sslHandshakeWithCertificateMessageIsNotLabeled() {
+        SSLHandshakeException tls = new SSLHandshakeException("PKIX path building failed: certificate chain invalid");
+        MongoSocketException e = new MongoSocketException("tls", ADDRESS, tls);
+        BackpressureErrorLabeler.applyLabelsIfEligible(e);
+        assertLacksBackpressureLabels(e);
+    }
+
+    @Test
+    void sslHandshakeWithoutConfigKeywordIsLabeled() {
+        SSLHandshakeException tls = new SSLHandshakeException("remote host closed connection during handshake");
+        MongoSocketException e = new MongoSocketException("tls", ADDRESS, tls);
+        BackpressureErrorLabeler.applyLabelsIfEligible(e);
+        assertHasBackpressureLabels(e);
+    }
+
+    @Test
+    void authErrorIsNotLabeled() {
+        MongoCredential credential = MongoCredential.createCredential("user", "db", "pwd".toCharArray());
+        MongoSecurityException e = new MongoSecurityException(credential, "auth failed");
+        BackpressureErrorLabeler.applyLabelsIfEligible(e);
+        assertLacksBackpressureLabels(e);
+    }
+
+    @Test
+    void commandExceptionIsNotLabeled() {
+        MongoException e = new MongoException(42, "some command error");
+        BackpressureErrorLabeler.applyLabelsIfEligible(e);
+        assertLacksBackpressureLabels(e);
+    }
+
+    @Test
+    void nonMongoThrowableIsNotLabeled() {
+        IOException e = new IOException("raw");
+        BackpressureErrorLabeler.applyLabelsIfEligible(e);
+    }
+
+    @Test
+    void sdamIssueTlsCheckDelegatesToHelper() {
+        MongoSocketException tlsException = new MongoSocketException("tls", ADDRESS, new CertificateException("bad cert"));
+        assertTrue(BackpressureErrorLabeler.isTlsConfigurationError(tlsException));
+
+        MongoSocketException plainException = new MongoSocketException("plain", ADDRESS);
+        assertFalse(BackpressureErrorLabeler.isTlsConfigurationError(plainException));
+
+        assertFalse(BackpressureErrorLabeler.isTlsConfigurationError(new MongoException(0, "not socket", new BsonDocument())));
+    }
+
+    private static void assertHasBackpressureLabels(final MongoException e) {
+        assertTrue(e.hasErrorLabel(MongoException.SYSTEM_OVERLOADED_ERROR_LABEL),
+                "expected SystemOverloadedError label");
+        assertTrue(e.hasErrorLabel(MongoException.RETRYABLE_ERROR_LABEL),
+                "expected RetryableError label");
+    }
+
+    private static void assertLacksBackpressureLabels(final MongoException e) {
+        assertFalse(e.hasErrorLabel(MongoException.SYSTEM_OVERLOADED_ERROR_LABEL),
+                "unexpected SystemOverloadedError label");
+        assertFalse(e.hasErrorLabel(MongoException.RETRYABLE_ERROR_LABEL),
+                "unexpected RetryableError label");
+    }
+}

driver-core/src/test/unit/com/mongodb/internal/connection/DefaultServerSpecification.groovy

@@ -258,6 +258,28 @@ class DefaultServerSpecification extends Specification {
         ]
     }
 
+    def 'DNS lookup failure should not be labeled as backpressure and should invalidate the pool'() {
+        given:
+        def exceptionToThrow = new MongoSocketException('DNS lookup failed', new ServerAddress(),
+                new UnknownHostException('no such host'))
+        BackpressureErrorLabeler.applyLabelsIfEligible(exceptionToThrow)
+        assert !exceptionToThrow.hasErrorLabel(MongoException.SYSTEM_OVERLOADED_ERROR_LABEL)
+
+        def connectionPool = Mock(ConnectionPool)
+        connectionPool.get(_) >> { throw exceptionToThrow }
+        def serverMonitor = Mock(ServerMonitor)
+        def server = defaultServer(connectionPool, serverMonitor)
+
+        when:
+        server.getConnection(createOperationContext())
+
+        then:
+        def e = thrown(MongoException)
+        e.is(exceptionToThrow)
+        1 * connectionPool.invalidate(exceptionToThrow)
+        1 * serverMonitor.cancelCurrentCheck()
+    }
+
     def 'failed authentication should invalidate the connection pool'() {
         given:
         def connectionPool = Mock(ConnectionPool)

driver-core/src/test/unit/com/mongodb/internal/connection/ServerDiscoveryAndMonitoringTest.java

@@ -54,9 +54,6 @@ public class ServerDiscoveryAndMonitoringTest extends AbstractServerDiscoveryAnd
 
     public ServerDiscoveryAndMonitoringTest(final String description, final BsonDocument definition) {
         super(definition);
-        assumeFalse("https://jira.mongodb.org/browse/JAVA-5949",
-                description.equals("error_handling_handshake.json: Network timeouts before and after the handshake completes"));
-
         this.description = description;
         init(serverAddress -> NO_OP_SERVER_LISTENER, NO_OP_CLUSTER_LISTENER);
     }