CSHARP-2988: Use white list for change stream resumability.

Author: DmitryLukyanov

src/MongoDB.Driver.Core/Core/Misc/Feature.cs

@@ -78,6 +78,7 @@ public class Feature
         private static readonly Feature __scramSha1Authentication = new Feature("ScramSha1Authentication", new SemanticVersion(3, 0, 0));
         private static readonly Feature __scramSha256Authentication = new Feature("ScramSha256Authentication", new SemanticVersion(4, 0, 0, ""));
         private static readonly Feature __serverExtractsUsernameFromX509Certificate = new Feature("ServerExtractsUsernameFromX509Certificate", new SemanticVersion(3, 3, 12));
+        private static readonly Feature __serverReturnsResumableChangeStreamErrorLabel = new Feature("ServerReturnsResumableChangeStreamErrorLabel", new SemanticVersion(4, 3, 0));
         private static readonly Feature __serverReturnsRetryableWriteErrorLabel = new Feature("ServerReturnsRetryableWriteErrorLabel", new SemanticVersion(4, 3, 0));
         private static readonly Feature __shardedTransactions = new Feature("ShardedTransactions", new SemanticVersion(4, 1, 6));
         private static readonly Feature __tailableCursor = new Feature("TailableCursor", new SemanticVersion(3, 2, 0));
@@ -362,6 +363,11 @@ public class Feature
         /// </summary>
         public static Feature ServerExtractsUsernameFromX509Certificate => __serverExtractsUsernameFromX509Certificate;
 
+        /// <summary>
+        /// Gets the server returns resumableChangeStream label feature.
+        /// </summary>
+        public static Feature ServerReturnsResumableChangeStreamErrorLabel => __serverReturnsResumableChangeStreamErrorLabel;
+
         /// <summary>
         /// Gets the server returns retryable writeError label feature.
         /// </summary>

src/MongoDB.Driver.Core/Core/Operations/ChangeStreamCursor.cs

@@ -45,6 +45,7 @@ internal sealed class ChangeStreamCursor<TDocument> : IChangeStreamCursor<TDocum
         private readonly BsonDocument _initialResumeAfter;
         private readonly BsonDocument _initialStartAfter;
         private readonly BsonTimestamp _initialStartAtOperationTime;
+        private readonly SemanticVersion _serverVersion;
 
         // public properties
         /// <inheritdoc />
@@ -63,6 +64,7 @@ internal sealed class ChangeStreamCursor<TDocument> : IChangeStreamCursor<TDocum
         /// <param name="initialStartAfter">The start after value.</param>
         /// <param name="initialResumeAfter">The resume after value.</param>
         /// <param name="initialStartAtOperationTime">The start at operation time value.</param>
+        /// <param name="serverVersion">The server version.</param>
         public ChangeStreamCursor(
             IAsyncCursor<RawBsonDocument> cursor,
             IBsonSerializer<TDocument> documentSerializer,
@@ -72,7 +74,8 @@ public ChangeStreamCursor(
             BsonTimestamp initialOperationTime,
             BsonDocument initialStartAfter,
             BsonDocument initialResumeAfter,
-            BsonTimestamp initialStartAtOperationTime)
+            BsonTimestamp initialStartAtOperationTime,
+            SemanticVersion serverVersion)
         {
             _cursor = Ensure.IsNotNull(cursor, nameof(cursor));
             _documentSerializer = Ensure.IsNotNull(documentSerializer, nameof(documentSerializer));
@@ -84,6 +87,7 @@ public ChangeStreamCursor(
             _initialStartAfter = initialStartAfter;
             _initialResumeAfter = initialResumeAfter;
             _initialStartAtOperationTime = initialStartAtOperationTime;
+            _serverVersion = Ensure.IsNotNull(serverVersion, nameof(serverVersion));
         }
 
         // public methods
@@ -119,7 +123,7 @@ public BsonDocument GetResumeToken()
                     hasMore = _cursor.MoveNext(cancellationToken);
                     break;
                 }
-                catch (Exception ex) when (RetryabilityHelper.IsResumableChangeStreamException(ex))
+                catch (Exception ex) when (RetryabilityHelper.IsResumableChangeStreamException(ex, _serverVersion))
                 {
                     var newCursor = Resume(cancellationToken);
                     _cursor.Dispose();
@@ -142,7 +146,7 @@ public BsonDocument GetResumeToken()
                     hasMore = await _cursor.MoveNextAsync(cancellationToken).ConfigureAwait(false);
                     break;
                 }
-                catch (Exception ex) when (RetryabilityHelper.IsResumableChangeStreamException(ex))
+                catch (Exception ex) when (RetryabilityHelper.IsResumableChangeStreamException(ex, _serverVersion))
                 {
                     var newCursor = await ResumeAsync(cancellationToken).ConfigureAwait(false);
                     _cursor.Dispose();

src/MongoDB.Driver.Core/Core/Operations/ChangeStreamOperation.cs

@@ -300,20 +300,21 @@ public IChangeStreamCursor<TResult> Execute(IReadBinding binding, CancellationTo
                 cursor = ExecuteAggregateOperation(context, cancellationToken);
                 cursorBatchInfo = (ICursorBatchInfo)cursor;
                 initialOperationTime = GetInitialOperationTimeIfRequired(context, cursorBatchInfo);
-            }
 
-            var postBatchResumeToken = GetInitialPostBatchResumeTokenIfRequired(cursorBatchInfo);
-
-            return new ChangeStreamCursor<TResult>(
-                cursor,
-                _resultSerializer,
-                bindingHandle.Fork(),
-                this,
-                postBatchResumeToken,
-                initialOperationTime,
-                _startAfter,
-                _resumeAfter,
-                _startAtOperationTime);
+                var postBatchResumeToken = GetInitialPostBatchResumeTokenIfRequired(cursorBatchInfo);
+
+                return new ChangeStreamCursor<TResult>(
+                    cursor,
+                    _resultSerializer,
+                    bindingHandle.Fork(),
+                    this,
+                    postBatchResumeToken,
+                    initialOperationTime,
+                    _startAfter,
+                    _resumeAfter,
+                    _startAtOperationTime,
+                    context.Channel.ConnectionDescription.ServerVersion);
+            }
         }
 
         /// <inheritdoc />
@@ -334,20 +335,21 @@ public async Task<IChangeStreamCursor<TResult>> ExecuteAsync(IReadBinding bindin
                 cursor = await ExecuteAggregateOperationAsync(context, cancellationToken).ConfigureAwait(false);
                 cursorBatchInfo = (ICursorBatchInfo)cursor;
                 initialOperationTime = GetInitialOperationTimeIfRequired(context, cursorBatchInfo);
-            }
 
-            var postBatchResumeToken = GetInitialPostBatchResumeTokenIfRequired(cursorBatchInfo);
-
-            return new ChangeStreamCursor<TResult>(
-                cursor,
-                _resultSerializer,
-                bindingHandle.Fork(),
-                this,
-                postBatchResumeToken,
-                initialOperationTime,
-                _startAfter,
-                _resumeAfter,
-                _startAtOperationTime);
+                var postBatchResumeToken = GetInitialPostBatchResumeTokenIfRequired(cursorBatchInfo);
+
+                return new ChangeStreamCursor<TResult>(
+                    cursor,
+                    _resultSerializer,
+                    bindingHandle.Fork(),
+                    this,
+                    postBatchResumeToken,
+                    initialOperationTime,
+                    _startAfter,
+                    _resumeAfter,
+                    _startAtOperationTime,
+                    context.Channel.ConnectionDescription.ServerVersion);
+            }
         }
 
         /// <inheritdoc />

src/MongoDB.Driver.Core/Core/Operations/RetryabilityHelper.cs

@@ -15,19 +15,18 @@
 
 using System;
 using System.Collections.Generic;
-using System.Linq;
-using MongoDB.Bson;
+using MongoDB.Driver.Core.Misc;
 
 namespace MongoDB.Driver.Core.Operations
 {
     internal static class RetryabilityHelper
     {
         // private constants
+        private const string ResumableChangeStreamErrorLabel = "ResumableChangeStreamError";
         private const string RetryableWriteErrorLabel = "RetryableWriteError";
 
         // private static fields
-        private static readonly HashSet<ServerErrorCode> __notResumableChangeStreamErrorCodes;
-        private static readonly HashSet<string> __notResumableChangeStreamErrorLabels;
+        private static readonly HashSet<ServerErrorCode> __resumableChangeStreamErrorCodes;
         private static readonly HashSet<Type> __resumableChangeStreamExceptions;
         private static readonly HashSet<Type> __retryableReadExceptions;
         private static readonly HashSet<Type> __retryableWriteExceptions;
@@ -39,15 +38,11 @@ static RetryabilityHelper()
         {
             var resumableAndRetryableExceptions = new HashSet<Type>()
             {
-                typeof(MongoConnectionException),
                 typeof(MongoNotPrimaryException),
                 typeof(MongoNodeIsRecoveringException)
             };
 
-            __resumableChangeStreamExceptions = new HashSet<Type>(resumableAndRetryableExceptions)
-            {
-                typeof(MongoCursorNotFoundException)
-            };
+            __resumableChangeStreamExceptions = new HashSet<Type>(resumableAndRetryableExceptions);
 
             __retryableReadExceptions = new HashSet<Type>(resumableAndRetryableExceptions);
 
@@ -68,16 +63,25 @@ static RetryabilityHelper()
                 ServerErrorCode.ExceededTimeLimit
             };
 
-            __notResumableChangeStreamErrorCodes = new HashSet<ServerErrorCode>()
-            {
-                ServerErrorCode.CappedPositionLost,
-                ServerErrorCode.CursorKilled,
-                ServerErrorCode.Interrupted
-            };
-
-            __notResumableChangeStreamErrorLabels = new HashSet<string>()
+            __resumableChangeStreamErrorCodes = new HashSet<ServerErrorCode>()
             {
-                "NonResumableChangeStreamError"
+                ServerErrorCode.HostUnreachable,
+                ServerErrorCode.HostNotFound,
+                ServerErrorCode.NetworkTimeout,
+                ServerErrorCode.ShutdownInProgress,
+                ServerErrorCode.PrimarySteppedDown,
+                ServerErrorCode.ExceededTimeLimit,
+                ServerErrorCode.SocketException,
+                ServerErrorCode.NotMaster,
+                ServerErrorCode.InterruptedAtShutdown,
+                ServerErrorCode.InterruptedDueToReplStateChange,
+                ServerErrorCode.NotMasterNoSlaveOk,
+                ServerErrorCode.NotMasterOrSecondary,
+                ServerErrorCode.StaleShardVersion,
+                ServerErrorCode.StaleEpoch,
+                ServerErrorCode.StaleConfig,
+                ServerErrorCode.RetryChangeStream,
+                ServerErrorCode.FailedToSatisfyReadPreference
             };
         }
 
@@ -90,26 +94,41 @@ public static void AddRetryableWriteErrorLabelIfRequired(MongoException exceptio
             }
         }
 
-        public static bool IsResumableChangeStreamException(Exception exception)
+        public static bool IsNetworkException(Exception exception)
         {
-            var commandException = exception as MongoCommandException;
-            if (commandException != null)
+            return exception is MongoConnectionException mongoConnectionException && mongoConnectionException.IsNetworkException;
+        }
+
+        public static bool IsResumableChangeStreamException(Exception exception, SemanticVersion serverVersion)
+        {
+            if (IsNetworkException(exception))
             {
-                var code = (ServerErrorCode)commandException.Code;
-                var isNonResumable =
-                    __notResumableChangeStreamErrorCodes.Contains(code) ||
-                    __notResumableChangeStreamErrorLabels.Any(c => commandException.HasErrorLabel(c));
-                return !isNonResumable;
+                return true;
+            }
+
+            if (Feature.ServerReturnsResumableChangeStreamErrorLabel.IsSupported(serverVersion))
+            {
+                return exception is MongoException mongoException ? mongoException.HasErrorLabel(ResumableChangeStreamErrorLabel) : false;
             }
             else
             {
+                var commandException = exception as MongoCommandException;
+                if (commandException != null)
+                {
+                    var code = (ServerErrorCode)commandException.Code;
+                    if (__resumableChangeStreamErrorCodes.Contains(code))
+                    {
+                        return true;
+                    }
+                }
+
                 return __resumableChangeStreamExceptions.Contains(exception.GetType());
             }
         }
 
         public static bool IsRetryableReadException(Exception exception)
         {
-            if (__retryableReadExceptions.Contains(exception.GetType()))
+            if (__retryableReadExceptions.Contains(exception.GetType()) || IsNetworkException(exception))
             {
                 return true;
             }
@@ -135,7 +154,7 @@ public static bool IsRetryableWriteException(Exception exception)
         // private static methods
         private static bool ShouldRetryableWriteExceptionLabelBeAdded(Exception exception)
         {
-            if (__retryableWriteExceptions.Contains(exception.GetType()))
+            if (__retryableWriteExceptions.Contains(exception.GetType()) || IsNetworkException(exception))
             {
                 return true;
             }

src/MongoDB.Driver.Core/MongoAuthenticationException.cs

@@ -62,5 +62,8 @@ public MongoAuthenticationException(SerializationInfo info, StreamingContext con
         {
         }
 #endif
+
+        /// <inheritdoc/>
+        public override bool IsNetworkException => false;
     }
 }

src/MongoDB.Driver.Core/MongoCommandException.cs

@@ -30,32 +30,6 @@ namespace MongoDB.Driver
 #endif
     public class MongoCommandException : MongoServerException
     {
-        #region static
-        /// <summary>
-        /// Adds error labels from a command result document into the top-level label list.
-        /// </summary>
-        /// <param name="exception">The exception.</param>
-        /// <param name="result">The result document.</param>
-        protected static void AddErrorLabelsFromCommandResult(MongoCommandException exception, BsonDocument result)
-        {
-            // note: make a best effort to extract the error labels from the result, but never throw an exception
-            if (result != null)
-            {
-                BsonValue errorLabels;
-                if (result.TryGetValue("errorLabels", out errorLabels) && errorLabels.IsBsonArray)
-                {
-                    foreach (var errorLabel in errorLabels.AsBsonArray)
-                    {
-                        if (errorLabel.IsString)
-                        {
-                            exception.AddErrorLabel(errorLabel.AsString);
-                        }
-                    }
-                }
-            }
-        }
-        #endregion
-
         // fields
         private readonly BsonDocument _command;
         private readonly BsonDocument _result;

src/MongoDB.Driver.Core/MongoConnectionClosedException.cs

@@ -50,5 +50,8 @@ protected MongoConnectionClosedException(SerializationInfo info, StreamingContex
         {
         }
 #endif
+
+        /// <inheritdoc/>
+        public override bool IsNetworkException => false;
     }
 }

src/MongoDB.Driver.Core/MongoConnectionException.cs

@@ -78,6 +78,11 @@ public ConnectionId ConnectionId
             get { return _connectionId; }
         }
 
+        /// <summary>
+        /// Determines whether the exception is network error or no.
+        /// </summary>
+        public virtual bool IsNetworkException => true; // true in subclasses, only if they can be considered as a network error
+
         // methods
 #if NET452
         /// <inheritdoc/>

src/MongoDB.Driver.Core/MongoExecutionTimeoutException.cs

@@ -76,6 +76,7 @@ public MongoExecutionTimeoutException(ConnectionId connectionId, string message,
             : base(connectionId, message, innerException)
         {
             _result = result;
+            AddErrorLabelsFromCommandResult(this, result);
         }
 
 #if NET452

src/MongoDB.Driver.Core/MongoServerException.cs

@@ -14,6 +14,7 @@
 */
 
 using System;
+using MongoDB.Bson;
 #if NET452
 using System.Runtime.Serialization;
 #endif
@@ -30,6 +31,32 @@ namespace MongoDB.Driver
 #endif
     public class MongoServerException : MongoException
     {
+        #region static
+        /// <summary>
+        /// Adds error labels from a command result document into the top-level label list.
+        /// </summary>
+        /// <param name="exception">The exception.</param>
+        /// <param name="result">The result document.</param>
+        protected static void AddErrorLabelsFromCommandResult(MongoServerException exception, BsonDocument result)
+        {
+            // note: make a best effort to extract the error labels from the result, but never throw an exception
+            if (result != null)
+            {
+                BsonValue errorLabels;
+                if (result.TryGetValue("errorLabels", out errorLabels) && errorLabels.IsBsonArray)
+                {
+                    foreach (var errorLabel in errorLabels.AsBsonArray)
+                    {
+                        if (errorLabel.IsString)
+                        {
+                            exception.AddErrorLabel(errorLabel.AsString);
+                        }
+                    }
+                }
+            }
+        }
+        #endregion
+
         // fields
         private readonly ConnectionId _connectionId;