Add baseBackoffMS support

.evergreen/.evg.yml

@@ -1740,6 +1740,10 @@ axes:
         display_name: "8.0"
         variables:
           VERSION: "8.0"
+      - id: "9.0"
+        display_name: "9.0"
+        variables:
+          VERSION: "9.0"
       # 8.2 is used solely for Windows testing. MongoDB 8.0 binaries are affected by SERVER-116018 on Windows,
       # and the fix is only available starting from 8.2.
       - id: "8.2"
@@ -2341,7 +2345,7 @@ buildvariants:
     #
 
   - matrix_name: "tests-zlib-compression"
-    matrix_spec: { compressor: "zlib", auth: "noauth", ssl: "nossl", jdk: "jdk8", version: [ "4.2", "4.4", "5.0", "6.0", "7.0", "8.0", "latest" ], topology: "standalone", os: "linux" }
+    matrix_spec: { compressor: "zlib", auth: "noauth", ssl: "nossl", jdk: "jdk8", version: [ "4.2", "4.4", "5.0", "6.0", "7.0", "8.0", "9.0", "latest" ], topology: "standalone", os: "linux" }
     display_name: "${version} ${compressor} ${topology} ${auth} ${ssl} ${jdk} ${os} "
     tags: [ "tests-variant" ]
     tasks:
@@ -2351,7 +2355,7 @@ buildvariants:
       - name: "test-legacy-task"
 
   - matrix_name: "tests-snappy-compression"
-    matrix_spec: { compressor: "snappy", auth: "noauth", ssl: "nossl", jdk: "jdk8", version: [ "4.2", "4.4", "5.0", "6.0", "7.0", "8.0", "latest" ], topology: "standalone", os: "linux" }
+    matrix_spec: { compressor: "snappy", auth: "noauth", ssl: "nossl", jdk: "jdk8", version: [ "4.2", "4.4", "5.0", "6.0", "7.0", "8.0", "9.0", "latest" ], topology: "standalone", os: "linux" }
     display_name: "${version} ${compressor} ${topology} ${auth} ${ssl} ${jdk} ${os} "
     tags: [ "tests-variant" ]
     tasks:
@@ -2362,7 +2366,7 @@ buildvariants:
 
   - matrix_name: "tests-zstd-compression"
     matrix_spec: { compressor: "zstd", auth: "noauth", ssl: "nossl", jdk: "jdk8",
-                   version: [ "4.2", "4.4", "5.0", "6.0", "7.0", "8.0", "latest" ],
+                   version: [ "4.2", "4.4", "5.0", "6.0", "7.0", "8.0", "9.0", "latest" ],
                    topology: "standalone", os: "linux" }
     display_name: "${version} ${compressor} ${topology} ${auth} ${ssl} ${jdk} ${os} "
     tags: [ "tests-variant" ]
@@ -2380,7 +2384,7 @@ buildvariants:
       - name: "test-bson-and-crypt-task"
 
   - matrix_name: "tests-jdk8-unsecure"
-    matrix_spec: { auth: "noauth", ssl: "nossl", jdk: "jdk8", version: [ "4.2", "4.4", "5.0", "6.0", "7.0", "8.0", "latest" ],
+    matrix_spec: { auth: "noauth", ssl: "nossl", jdk: "jdk8", version: [ "4.2", "4.4", "5.0", "6.0", "7.0", "8.0", "9.0", "latest" ],
                    topology: "*", os: "linux" }
     display_name: "${version} ${topology} ${auth} ${ssl} ${jdk} ${os} "
     tags: [ "tests-variant" ]
@@ -2392,7 +2396,7 @@ buildvariants:
 
   - matrix_name: "tests-jdk-secure"
     matrix_spec: { auth: "auth", ssl: "ssl", jdk: [ "jdk8", "jdk17", "jdk21" ],
-                   version: [ "4.2", "4.4", "5.0", "6.0", "7.0", "8.0", "latest" ],
+                   version: [ "4.2", "4.4", "5.0", "6.0", "7.0", "8.0", "9.0", "latest" ],
                    topology: "*", os: "linux" }
     display_name: "${version} ${topology} ${auth} ${ssl} ${jdk} ${os} "
     tags: [ "tests-variant" ]
@@ -2429,7 +2433,7 @@ buildvariants:
         exec_timeout_secs: 7200
 
   - matrix_name: "tests-require-api-version"
-    matrix_spec: { api-version: "required", auth: "auth", ssl: "nossl", jdk: [ "jdk21" ], version: [ "5.0", "6.0", "7.0", "8.0", "latest" ],
+    matrix_spec: { api-version: "required", auth: "auth", ssl: "nossl", jdk: [ "jdk21" ], version: [ "5.0", "6.0", "7.0", "8.0", "9.0", "latest" ],
                    topology: "standalone", os: "linux" }
     display_name: "${version} ${topology} ${api-version} "
     tags: [ "tests-variant" ]
@@ -2440,7 +2444,7 @@ buildvariants:
       - name: "test-legacy-task"
 
   - matrix_name: "tests-load-balancer-secure"
-    matrix_spec: { auth: "auth", ssl: "ssl", jdk: [ "jdk21" ], version: [ "5.0", "6.0", "7.0", "8.0", "latest" ], topology: "sharded-cluster",
+    matrix_spec: { auth: "auth", ssl: "ssl", jdk: [ "jdk21" ], version: [ "5.0", "6.0", "7.0", "8.0", "9.0", "latest" ], topology: "sharded-cluster",
                    os: "ubuntu" }
     display_name: "Load Balancer ${version} ${auth} ${ssl} ${jdk} ${os}"
     tasks:
@@ -2500,7 +2504,7 @@ buildvariants:
       - name: "gssapi-auth-test-task"
 
   - matrix_name: "aws-auth-test"
-    matrix_spec: { ssl: "nossl", jdk: [ "jdk8", "jdk17", "jdk21" ], version: [ "4.4", "5.0", "6.0", "7.0", "8.0", "latest" ], os: "ubuntu",
+    matrix_spec: { ssl: "nossl", jdk: [ "jdk8", "jdk17", "jdk21" ], version: [ "4.4", "5.0", "6.0", "7.0", "8.0", "9.0", "latest" ], os: "ubuntu",
                    aws-credential-provider: "*" }
     display_name: "MONGODB-AWS Basic Auth test ${version} ${jdk} ${aws-credential-provider}"
     run_on: "ubuntu2204-small"
@@ -2519,15 +2523,15 @@ buildvariants:
       - name: "aws-auth-test-with-web-identity-credentials-task"
 
   - matrix_name: "accept-api-version-2-test"
-    matrix_spec: { ssl: "nossl", auth: "noauth", jdk: "jdk21", version: [ "5.0", "6.0", "7.0", "8.0", "latest" ], topology: "standalone",
+    matrix_spec: { ssl: "nossl", auth: "noauth", jdk: "jdk21", version: [ "5.0", "6.0", "7.0", "8.0", "9.0", "latest" ], topology: "standalone",
                    os: "linux" }
     display_name: "Accept API Version 2 ${version}"
     run_on: "ubuntu2204-small"
     tasks:
       - name: "accept-api-version-2-test-task"
 
   - matrix_name: "ocsp-test"
-    matrix_spec: { auth: "noauth", ssl: "ssl", jdk: "jdk21", version: [ "4.4", "5.0", "6.0", "7.0", "8.0", "latest" ], os: "ubuntu" }
+    matrix_spec: { auth: "noauth", ssl: "ssl", jdk: "jdk21", version: [ "4.4", "5.0", "6.0", "7.0", "8.0", "9.0", "latest" ], os: "ubuntu" }
     display_name: "OCSP test ${version} ${os}"
     tasks:
       - name: ".ocsp"
@@ -2560,14 +2564,14 @@ buildvariants:
       - name: ".csfle-aws-from-environment"
 
   - matrix_name: "csfle-tests-with-mongocryptd"
-    matrix_spec: { os: "linux", version: [ "4.2", "4.4", "5.0", "6.0", "7.0", "8.0", "latest" ], topology: [ "replicaset" ] }
+    matrix_spec: { os: "linux", version: [ "4.2", "4.4", "5.0", "6.0", "7.0", "8.0", "9.0", "latest" ], topology: [ "replicaset" ] }
     display_name: "CSFLE with mongocryptd: ${version}"
     tasks:
       - name: "csfle-tests-with-mongocryptd-task"
 
   - matrix_name: "csfle-tests-with-mongocryptd-windows"
     matrix_spec: { os: "windows",
-                   version: [ "8.0", "latest" ],
+                   version: [ "8.0", "9.0", "latest" ],
                    topology: [ "replicaset" ] }
     display_name: "${os} CSFLE with mongocryptd: ${version}"
     tasks:

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

@@ -173,7 +173,7 @@ private InternalConnectionInitializationDescription createInitializationDescript
     private BsonDocument createHelloCommand(final Authenticator authenticator, final InternalConnection connection) {
         BsonDocument helloCommandDocument = new BsonDocument(getHandshakeCommandName(), new BsonInt32(1))
                 .append("helloOk", BsonBoolean.TRUE)
-                .append("backpressure", BsonBoolean.TRUE);
+                .append("backpressure", new BsonString("2"));
         if (clientMetadataDocument != null) {
             helloCommandDocument.append("client", clientMetadataDocument);
         }

driver-core/src/main/com/mongodb/internal/operation/AsyncOperationHelper.java

@@ -58,7 +58,6 @@
 import static com.mongodb.internal.operation.CommandOperationHelper.createSpecRetryControl;
 import static com.mongodb.internal.operation.CommandOperationHelper.transformWriteException;
 import static com.mongodb.internal.operation.CommandOperationHelper.isWriteRetryRequirementsMet;
-import static com.mongodb.internal.operation.OperationHelper.isServerWriteRetryRequirementsMet;
 import static com.mongodb.internal.operation.WriteConcernHelper.throwOnWriteConcernError;
 
 final class AsyncOperationHelper {
@@ -306,7 +305,7 @@ static <T, R> void executeRetryableWriteAsync(
                             (source, connection, operationContextWithMinRtt, functionCallback) -> {
                                 beginAsync().<T>thenSupply(executeCommandCallback -> {
                                     ConnectionDescription connectionDescription = connection.getDescription();
-                                    retryControl.breakAndThrowIfRetryAnd(() -> !isServerWriteRetryRequirementsMet(connectionDescription));
+                                    retryControl.breakAndThrowIfRetryAnd(() -> retryControl.getPolicy().shouldBreakWriteLoop(connectionDescription));
                                     if (command.getNullable() == null) {
                                         command.set(commandCreator.create(operationContextWithMinRtt, source.getServerDescription(), connectionDescription));
                                     } else {

driver-core/src/main/com/mongodb/internal/operation/ClientBulkWriteOperation.java

@@ -131,7 +131,6 @@
 import static com.mongodb.internal.operation.CommandOperationHelper.createSpecRetryControl;
 import static com.mongodb.internal.operation.CommandOperationHelper.transformWriteException;
 import static com.mongodb.internal.operation.OperationHelper.isNonCommandWriteRetryRequirementsMet;
-import static com.mongodb.internal.operation.OperationHelper.isServerWriteRetryRequirementsMet;
 import static com.mongodb.internal.operation.SyncOperationHelper.cursorDocumentToBatchCursor;
 import static com.mongodb.internal.operation.SyncOperationHelper.decorateWithRetries;
 import static com.mongodb.internal.operation.SyncOperationHelper.withSourceAndConnection;
@@ -305,7 +304,7 @@ private Integer executeBatch(
                         (connectionSource, connection, operationContextWithMinRtt) -> {
                             retryControl.getPolicy().onCommand(() -> BULK_WRITE_COMMAND_NAME);
                             ConnectionDescription connectionDescription = connection.getDescription();
-                            retryControl.breakAndThrowIfRetryAnd(() -> !isServerWriteRetryRequirementsMet(connectionDescription));
+                            retryControl.breakAndThrowIfRetryAnd(() -> retryControl.getPolicy().shouldBreakWriteLoop(connectionDescription));
                             resultAccumulator.onNewServerAddress(connectionDescription.getServerAddress());
                             ClientBulkWriteCommand bulkWriteCommand = createBulkWriteCommand(
                                     retryControl, connectionDescription, effectiveWriteConcern, sessionContext, unexecutedModels, batchEncoder);
@@ -360,7 +359,7 @@ private void executeBatchAsync(
                                 beginAsync().<ExhaustiveClientBulkWriteCommandOkResponse>thenSupply(executeAndExhaustCallback -> {
                                     retryControl.getPolicy().onCommand(() -> BULK_WRITE_COMMAND_NAME);
                                     ConnectionDescription connectionDescription = connection.getDescription();
-                                    retryControl.breakAndThrowIfRetryAnd(() -> !isServerWriteRetryRequirementsMet(connectionDescription));
+                                    retryControl.breakAndThrowIfRetryAnd(() -> retryControl.getPolicy().shouldBreakWriteLoop(connectionDescription));
                                     resultAccumulator.onNewServerAddress(connectionDescription.getServerAddress());
                                     ClientBulkWriteCommand bulkWriteCommand = createBulkWriteCommand(
                                             retryControl, connectionDescription, effectiveWriteConcern, sessionContext, unexecutedModels, batchEncoder);

driver-core/src/main/com/mongodb/internal/operation/MixedBulkWriteOperation.java

@@ -63,7 +63,6 @@
 import static com.mongodb.internal.operation.CommandOperationHelper.createSpecRetryControl;
 import static com.mongodb.internal.operation.CommandOperationHelper.transformWriteException;
 import static com.mongodb.internal.operation.CommandOperationHelper.validateAndGetEffectiveWriteConcern;
-import static com.mongodb.internal.operation.OperationHelper.isServerWriteRetryRequirementsMet;
 import static com.mongodb.internal.operation.OperationHelper.validateWriteRequests;
 import static com.mongodb.internal.operation.SyncOperationHelper.decorateWithRetries;
 import static com.mongodb.internal.operation.SyncOperationHelper.withSourceAndConnection;
@@ -416,7 +415,7 @@ private void onNewConnection(
             final ConnectionDescription connectionDescription,
             final WriteConcern effectiveWriteConcern,
             final RetryControl<SpecRetryPolicy> retryControl) {
-        retryControl.breakAndThrowIfRetryAnd(() -> !isServerWriteRetryRequirementsMet(connectionDescription));
+        retryControl.breakAndThrowIfRetryAnd(() -> retryControl.getPolicy().shouldBreakWriteLoop(connectionDescription));
         validateWriteRequests(connectionDescription, bypassDocumentValidation, writeRequests, effectiveWriteConcern);
     }
 

driver-core/src/main/com/mongodb/internal/operation/SpecRetryPolicy.java

@@ -16,6 +16,7 @@
 package com.mongodb.internal.operation;
 
 import com.mongodb.MongoClientSettings;
+import com.mongodb.MongoCommandException;
 import com.mongodb.MongoConnectionPoolClearedException;
 import com.mongodb.MongoException;
 import com.mongodb.MongoOperationTimeoutException;
@@ -33,6 +34,9 @@
 import com.mongodb.internal.time.ExponentialBackoff;
 import com.mongodb.lang.Nullable;
 
+import org.bson.BsonDocument;
+import org.bson.BsonValue;
+
 import java.time.Duration;
 import java.util.EnumMap;
 import java.util.EnumSet;
@@ -55,6 +59,7 @@
 import static com.mongodb.internal.operation.CommandOperationHelper.isRetryableException;
 import static com.mongodb.internal.operation.OperationHelper.LOGGER;
 import static com.mongodb.internal.operation.OperationHelper.isReadRetryRequirementsMet;
+import static com.mongodb.internal.operation.OperationHelper.isServerWriteRetryRequirementsMet;
 import static java.lang.Boolean.TRUE;
 import static java.lang.String.format;
 import static java.util.Arrays.asList;
@@ -66,6 +71,7 @@
  */
 final class SpecRetryPolicy implements RetryPolicy {
     private static final int INFINITE_ATTEMPTS = Integer.MAX_VALUE;
+    private static final String BASE_BACKOFF_MS_FIELD = "baseBackoffMS";
 
     private final IndividualPolicies policies;
     private int maxAttempts;
@@ -285,11 +291,48 @@ private static Duration calculateOverloadBackoff(final Throwable attemptFailedRe
         assertFalse(attemptFailedResult instanceof OperationHelper.ResourceSupplierInternalException);
         if (attemptFailedResult instanceof MongoException
                 && ((MongoException) attemptFailedResult).hasErrorLabel(SYSTEM_OVERLOADED_ERROR_LABEL)) {
-            return ExponentialBackoff.calculateOverloadBackoff(immediateNextAttempt);
+            return ExponentialBackoff.calculateOverloadBackoff(immediateNextAttempt, extractBaseBackoffMs(attemptFailedResult));
         }
         return Duration.ZERO;
     }
 
+    @Nullable
+    private static Long extractBaseBackoffMs(final Throwable attemptFailedResult) {
+        if (!(attemptFailedResult instanceof MongoCommandException)) {
+            return null;
+        }
+        BsonDocument response = ((MongoCommandException) attemptFailedResult).getResponse();
+        if (!response.containsKey(BASE_BACKOFF_MS_FIELD)) {
+            return null;
+        }
+        BsonValue value = response.get(BASE_BACKOFF_MS_FIELD);
+        if (!value.isNumber()) {
+            return null;
+        }
+        long parsed = value.asNumber().longValue();
+        return parsed > 0 ? parsed : null;
+    }
+
+    /**
+     * Decides whether the write retry loop should be broken before the next attempt.
+     * The loop is not broken when all failures observed so far within the current command execution
+     * are retryable overload errors, because such commands were load-shed by the server without
+     * being executed, making the overload retry policy independent of retryable-write server support.
+     * Otherwise, the loop is broken when the server does not support retryable writes,
+     * preserving the existing retryable-write behavior.
+     *
+     * @param connectionDescription The {@link ConnectionDescription} of the connection selected for
+     *                              the immediate next attempt.
+     * @return {@code true} iff the write retry loop must be broken and the prospective failed result thrown.
+     */
+    boolean shouldBreakWriteLoop(final ConnectionDescription connectionDescription) {
+        assertTrue(policies.write().isPresent());
+        if (policies.overload().map(IndividualPolicies.State.Overload::observedNoneOrOnlyRetryableOverloadErrors).orElse(false)) {
+            return false;
+        }
+        return !isServerWriteRetryRequirementsMet(connectionDescription);
+    }
+
     private static int maxAttempts(final int maxRetries) {
         if (maxRetries == INFINITE_ATTEMPTS) {
             return INFINITE_ATTEMPTS;
@@ -570,6 +613,14 @@ static final class Overload extends State {
                 private final ErrorPropagation errorPropagation;
                 @Nullable
                 private BaseClientSessionImpl.OverloadRetryPolicyState sessionScopedState;
+                /**
+                 * Whether all failures observed so far within the current command execution are
+                 * retryable overload errors. {@code true} is also the case if no failures have been observed.
+                 *
+                 * @see MongoException#RETRYABLE_ERROR_LABEL
+                 * @see MongoException#SYSTEM_OVERLOADED_ERROR_LABEL
+                 */
+                private boolean observedNoneOrOnlyRetryableOverloadErrors;
 
                 Overload(
                         final boolean effectiveRetrySetting,
@@ -581,6 +632,7 @@ static final class Overload extends State {
                     this.maxAdaptiveRetriesSetting = maxAdaptiveRetriesSetting;
                     this.errorPropagation = errorPropagation;
                     sessionScopedState = null;
+                    observedNoneOrOnlyRetryableOverloadErrors = true;
                 }
 
                 @Nullable
@@ -611,7 +663,16 @@ LabelInfo onAttemptFailure(final Throwable exception) {
                     return labelInfo;
                 }
 
+                /**
+                 * Returns whether all failures observed so far within the current command execution are
+                 * retryable overload errors. {@code true} is also returned if no failures have been observed.
+                 */
+                boolean observedNoneOrOnlyRetryableOverloadErrors() {
+                    return observedNoneOrOnlyRetryableOverloadErrors;
+                }
+
                 private void onAnyAttemptFailure(final boolean retryableOverloadError) {
+                    observedNoneOrOnlyRetryableOverloadErrors &= retryableOverloadError;
                     BaseClientSessionImpl.OverloadRetryPolicyState localSessionScopedState = assertNotNull(sessionScopedState);
                     assertNotNull(localSessionScopedState.getCommandExecutionScoped()).onAnyAttemptFailure(retryableOverloadError);
                     BaseClientSessionImpl.OverloadRetryPolicyState.CommitScoped commitScopedState = localSessionScopedState.getCommitScoped();
@@ -638,6 +699,7 @@ public String toString() {
                             + ", maxAdaptiveRetriesSetting=" + maxAdaptiveRetriesSetting
                             + ", errorPropagation=" + errorPropagation
                             + ", sessionScopedState=" + sessionScopedState
+                            + ", observedNoneOrOnlyRetryableOverloadErrors=" + observedNoneOrOnlyRetryableOverloadErrors
                             + '}';
                 }