CSHARP-3156: Treat CursorNotFound as a resumable change stream error

Author: MikalaiMazurenka

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

@@ -273,39 +273,55 @@ private BsonDocument CreateKillCursorsCommand()
         private CursorBatch<TDocument> ExecuteGetMoreCommand(IChannelHandle channel, CancellationToken cancellationToken)
         {
             var command = CreateGetMoreCommand();
-            var result = channel.Command<BsonDocument>(
-                _channelSource.Session,
-                null, // readPreference
-                _collectionNamespace.DatabaseNamespace,
-                command,
-                null, // commandPayloads
-                NoOpElementNameValidator.Instance,
-                null, // additionalOptions
-                null, // postWriteAction
-                CommandResponseHandling.Return,
-                __getMoreCommandResultSerializer,
-                _messageEncoderSettings,
-                cancellationToken);
+            BsonDocument result;
+            try
+            {
+                result = channel.Command<BsonDocument>(
+                    _channelSource.Session,
+                    null, // readPreference
+                    _collectionNamespace.DatabaseNamespace,
+                    command,
+                    null, // commandPayloads
+                    NoOpElementNameValidator.Instance,
+                    null, // additionalOptions
+                    null, // postWriteAction
+                    CommandResponseHandling.Return,
+                    __getMoreCommandResultSerializer,
+                    _messageEncoderSettings,
+                    cancellationToken);
+            }
+            catch (MongoCommandException ex) when (IsMongoCursorNotFoundException(ex))
+            {
+                throw new MongoCursorNotFoundException(channel.ConnectionDescription.ConnectionId, _cursorId, command);
+            }
 
             return CreateCursorBatch(result);
         }
 
         private async Task<CursorBatch<TDocument>> ExecuteGetMoreCommandAsync(IChannelHandle channel, CancellationToken cancellationToken)
         {
             var command = CreateGetMoreCommand();
-            var result = await channel.CommandAsync<BsonDocument>(
-                _channelSource.Session,
-                null, // readPreference
-                _collectionNamespace.DatabaseNamespace,
-                command,
-                null, // commandPayloads
-                NoOpElementNameValidator.Instance,
-                null, // additionalOptions
-                null, // postWriteAction
-                CommandResponseHandling.Return,
-                __getMoreCommandResultSerializer,
-                _messageEncoderSettings,
-                cancellationToken).ConfigureAwait(false);
+            BsonDocument result;
+            try
+            {
+                result = await channel.CommandAsync<BsonDocument>(
+                    _channelSource.Session,
+                    null, // readPreference
+                    _collectionNamespace.DatabaseNamespace,
+                    command,
+                    null, // commandPayloads
+                    NoOpElementNameValidator.Instance,
+                    null, // additionalOptions
+                    null, // postWriteAction
+                    CommandResponseHandling.Return,
+                    __getMoreCommandResultSerializer,
+                    _messageEncoderSettings,
+                    cancellationToken).ConfigureAwait(false);
+            }
+            catch (MongoCommandException ex) when (IsMongoCursorNotFoundException(ex))
+            {
+                throw new MongoCursorNotFoundException(channel.ConnectionDescription.ConnectionId, _cursorId, command);
+            }
 
             return CreateCursorBatch(result);
         }
@@ -515,6 +531,11 @@ private async Task<CursorBatch<TDocument>> GetNextBatchAsync(CancellationToken c
             }
         }
 
+        private bool IsMongoCursorNotFoundException(MongoCommandException exception)
+        {
+            return exception.Code == (int)ServerErrorCode.CursorNotFound;
+        }
+
         private void KillCursors(CancellationToken cancellationToken)
         {
             using (EventContext.BeginOperation(_operationId))

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

@@ -109,6 +109,10 @@ public static bool IsResumableChangeStreamException(Exception exception, Semanti
             {
                 return true;
             }
+            if (exception is MongoCursorNotFoundException)
+            {
+                return true;
+            }
 
             if (Feature.ServerReturnsResumableChangeStreamErrorLabel.IsSupported(serverVersion))
             {

src/MongoDB.Driver.Core/ServerErrorCode.cs

@@ -21,6 +21,7 @@ internal enum ServerErrorCode
         // see: https://github.com/mongodb/mongo/blob/master/src/mongo/base/error_codes.yml
         CappedPositionLost = 136,
         CursorKilled = 237,
+        CursorNotFound = 43,
         ElectionInProgress = 216,
         ExceededTimeLimit = 262,
         FailedToSatisfyReadPreference = 133,

tests/MongoDB.Driver.Core.Tests/Core/Operations/RetryabilityHelperTests.cs

@@ -135,7 +135,7 @@ public void IsCommandRetryable_should_return_expected_result(string command, boo
         [InlineData(typeof(MongoConnectionClosedException), false)]
         [InlineData(typeof(MongoNodeIsRecoveringException), true)]
         [InlineData(typeof(MongoNotPrimaryException), true)]
-        [InlineData(typeof(MongoCursorNotFoundException), false)]
+        [InlineData(typeof(MongoCursorNotFoundException), true)]
         [InlineData(ServerErrorCode.HostNotFound, true)]
         [InlineData(ServerErrorCode.HostUnreachable, true)]
         [InlineData(ServerErrorCode.NetworkTimeout, true)]
@@ -193,9 +193,10 @@ public void IsResumableChangeStreamException_should_return_expected_result_for_s
         [Theory]
         [InlineData(typeof(MongoConnectionException), true)] // network exception
         [InlineData(typeof(MongoConnectionClosedException), false)]
-        public void IsResumableChangeStreamException_should_return_expected_result_for_servers_with_new_behavior_and_connection_errors(Type exceptionType, bool isResumable)
+        [InlineData(typeof(MongoCursorNotFoundException), true)]
+        public void IsResumableChangeStreamException_should_return_expected_result_for_servers_with_new_behavior_and_errors(Type exceptionType, bool isResumable)
         {
-            var exception = (MongoConnectionException)CoreExceptionHelper.CreateException(exceptionType);
+            var exception = (MongoException)CoreExceptionHelper.CreateException(exceptionType);
 
             var result = RetryabilityHelper.IsResumableChangeStreamException(exception, Feature.ServerReturnsResumableChangeStreamErrorLabel.FirstSupportedVersion);
 

tests/MongoDB.Driver.Tests/Specifications/change-streams/tests/change-streams-errors.json

@@ -102,15 +102,12 @@
       ],
       "result": {
         "error": {
-          "code": 280,
-          "errorLabels": [
-            "NonResumableChangeStreamError"
-          ]
+          "code": 280
         }
       }
     },
     {
-      "description": "change stream errors on MaxTimeMSExpired",
+      "description": "change stream errors on ElectionInProgress",
       "minServerVersion": "4.2",
       "failPoint": {
         "configureFailPoint": "failCommand",
@@ -121,7 +118,7 @@
           "failCommands": [
             "getMore"
           ],
-          "errorCode": 50,
+          "errorCode": 216,
           "closeConnection": false
         }
       },
@@ -130,13 +127,7 @@
         "replicaset",
         "sharded"
       ],
-      "changeStreamPipeline": [
-        {
-          "$project": {
-            "_id": 0
-          }
-        }
-      ],
+      "changeStreamPipeline": [],
       "changeStreamOptions": {},
       "operations": [
         {
@@ -152,7 +143,7 @@
       ],
       "result": {
         "error": {
-          "code": 50
+          "code": 216
         }
       }
     }

tests/MongoDB.Driver.Tests/Specifications/change-streams/tests/change-streams-errors.yml

@@ -72,24 +72,21 @@ tests:
     result:
       error:
         code: 280
-        errorLabels: [ "NonResumableChangeStreamError" ]
   -
-    description: change stream errors on MaxTimeMSExpired
+    description: change stream errors on ElectionInProgress
     minServerVersion: "4.2"
     failPoint:
       configureFailPoint: failCommand
       mode: { times: 1 }
       data:
           failCommands: ["getMore"]
-          errorCode: 50 # An error code that's not on the old blacklist or whitelist
+          errorCode: 216 # An error code that's not on the old blacklist or whitelist
           closeConnection: false
     target: collection
     topology:
       - replicaset
       - sharded
-    changeStreamPipeline:
-      -
-        $project: { _id: 0 }
+    changeStreamPipeline: []
     changeStreamOptions: {}
     operations:
       -
@@ -101,4 +98,4 @@ tests:
             z: 3
     result:
       error:
-        code: 50
+        code: 216

tests/MongoDB.Driver.Tests/Specifications/change-streams/tests/change-streams-resume-whitelist.json

@@ -1648,6 +1648,102 @@
           }
         ]
       }
+    },
+    {
+      "description": "change stream resumes after CursorNotFound",
+      "minServerVersion": "4.2",
+      "failPoint": {
+        "configureFailPoint": "failCommand",
+        "mode": {
+          "times": 1
+        },
+        "data": {
+          "failCommands": [
+            "getMore"
+          ],
+          "errorCode": 43,
+          "closeConnection": false
+        }
+      },
+      "target": "collection",
+      "topology": [
+        "replicaset",
+        "sharded"
+      ],
+      "changeStreamPipeline": [],
+      "changeStreamOptions": {},
+      "operations": [
+        {
+          "database": "change-stream-tests",
+          "collection": "test",
+          "name": "insertOne",
+          "arguments": {
+            "document": {
+              "x": 1
+            }
+          }
+        }
+      ],
+      "expectations": [
+        {
+          "command_started_event": {
+            "command": {
+              "aggregate": "test",
+              "cursor": {},
+              "pipeline": [
+                {
+                  "$changeStream": {}
+                }
+              ]
+            },
+            "command_name": "aggregate",
+            "database_name": "change-stream-tests"
+          }
+        },
+        {
+          "command_started_event": {
+            "command": {
+              "getMore": 42,
+              "collection": "test"
+            },
+            "command_name": "getMore",
+            "database_name": "change-stream-tests"
+          }
+        },
+        {
+          "command_started_event": {
+            "command": {
+              "aggregate": "test",
+              "cursor": {},
+              "pipeline": [
+                {
+                  "$changeStream": {}
+                }
+              ]
+            },
+            "command_name": "aggregate",
+            "database_name": "change-stream-tests"
+          }
+        }
+      ],
+      "result": {
+        "success": [
+          {
+            "_id": "42",
+            "documentKey": "42",
+            "operationType": "insert",
+            "ns": {
+              "db": "change-stream-tests",
+              "coll": "test"
+            },
+            "fullDocument": {
+              "x": {
+                "$numberInt": "1"
+              }
+            }
+          }
+        ]
+      }
     }
   ]
 }

tests/MongoDB.Driver.Tests/Specifications/change-streams/tests/change-streams-resume-whitelist.yml

@@ -1105,3 +1105,69 @@ tests:
           fullDocument:
             x:
               $numberInt: "1"
+  -
+    # CursorNotFound is special-cased to be resumable regardless of server versions or error labels, so this test has
+    # no maxWireVersion.
+    description: "change stream resumes after CursorNotFound"
+    minServerVersion: "4.2"
+    failPoint:
+      configureFailPoint: failCommand
+      mode: { times: 1 }
+      data:
+          failCommands: ["getMore"]
+          errorCode: 43
+          closeConnection: false
+    target: collection
+    topology:
+      - replicaset
+      - sharded
+    changeStreamPipeline: []
+    changeStreamOptions: {}
+    operations:
+      -
+        database: *database_name
+        collection: *collection_name
+        name: insertOne
+        arguments:
+          document:
+            x: 1
+    expectations:
+      -
+        command_started_event:
+          command:
+            aggregate: *collection_name
+            cursor: {}
+            pipeline:
+              -
+                $changeStream: {}
+          command_name: aggregate
+          database_name: *database_name
+      -
+        command_started_event:
+          command:
+            getMore: 42
+            collection: *collection_name
+          command_name: getMore
+          database_name: *database_name
+      -
+        command_started_event:
+          command:
+            aggregate: *collection_name
+            cursor: {}
+            pipeline:
+              -
+                $changeStream: {}
+          command_name: aggregate
+          database_name: *database_name
+    result:
+      success:
+        -
+          _id: "42"
+          documentKey: "42"
+          operationType: insert
+          ns:
+            db: *database_name
+            coll: *collection_name
+          fullDocument:
+            x:
+              $numberInt: "1"