GODRIVER-1678 Consider CursorNotFound a resumable change stream error (#445)

Divjot Arora · web-flow · commit 4979f1604278 · 2020-07-09T18:05:44.000-04:00
This commit also resyncs all change stream spec tests to pull in changes from GODRIVER-1671.
diff --git a/data/change-streams/README.rst b/data/change-streams/README.rst
@@ -63,7 +63,7 @@ The definition of MATCH or MATCHES in the Spec Test Runner is as follows:
 
 - MATCH takes two values, ``expected`` and ``actual``
 - Notation is "Assert [actual] MATCHES [expected]
-- Assertion passes if ``expected`` is a subset of ``actual``, with the values ``42`` and ``"42"`` acting as placeholders for "any value"
+- Assertion passes if ``expected`` is a subset of ``actual``, with the value ``42`` acting as placeholders for "any value"
 
 Pseudocode implementation of ``actual`` MATCHES ``expected``:
 
@@ -149,6 +149,20 @@ Although synchronous drivers must provide a `non-blocking mode of iteration <../
 
 If the test expects an error and one was not thrown by either creating the change stream or executing the test's operations, iterating the change stream once allows for an error to be thrown by a ``getMore`` command. If the test does not expect any error, the change stream should be iterated only until it returns as many result documents as are expected by the test.
 
+Testing on Sharded Clusters
+---------------------------
+
+When writing data on sharded clusters, majority-committed data does not always show up in the response of the first
+``getMore`` command after the data is written. This is because in sharded clusters, no data from shard A may be returned
+until all other shard reports an entry that sorts after the change in shard A.
+
+To account for this, drivers MUST NOT rely on change stream documents in certain batches. For example, if expecting two
+documents in a change stream, these may not be part of the same ``getMore`` response, or even be produced in two
+subsequent ``getMore`` responses. Drivers MUST allow for a ``getMore`` to produce empty batches when testing on a
+sharded cluster. By default, this can take up to 10 seconds, but can be controlled by enabling the ``writePeriodicNoops``
+server parameter and configuring the ``periodNoopIntervalSecs`` parameter. Choosing lower values allows for running
+change stream tests with smaller timeouts.
+
 Prose Tests
 ===========
 
diff --git a/data/change-streams/change-streams-errors.json b/data/change-streams/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
         }
       }
     }
diff --git a/data/change-streams/change-streams-errors.yml b/data/change-streams/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
diff --git a/data/change-streams/change-streams-resume-whitelist.json b/data/change-streams/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"
+              }
+            }
+          }
+        ]
+      }
     }
   ]
 }
diff --git a/data/change-streams/change-streams-resume-whitelist.yml b/data/change-streams/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"
diff --git a/data/change-streams/change-streams.json b/data/change-streams/change-streams.json
@@ -33,9 +33,7 @@
               "cursor": {},
               "pipeline": [
                 {
-                  "$changeStream": {
-                    "fullDocument": "default"
-                  }
+                  "$changeStream": {}
                 }
               ]
             },
@@ -153,9 +151,7 @@
               "cursor": {},
               "pipeline": [
                 {
-                  "$changeStream": {
-                    "fullDocument": "default"
-                  }
+                  "$changeStream": {}
                 }
               ]
             },
@@ -226,9 +222,7 @@
               "cursor": {},
               "pipeline": [
                 {
-                  "$changeStream": {
-                    "fullDocument": "default"
-                  }
+                  "$changeStream": {}
                 },
                 {
                   "$match": {
@@ -312,9 +306,7 @@
               "cursor": {},
               "pipeline": [
                 {
-                  "$changeStream": {
-                    "fullDocument": "default"
-                  }
+                  "$changeStream": {}
                 }
               ]
             },
@@ -404,7 +396,6 @@
               "pipeline": [
                 {
                   "$changeStream": {
-                    "fullDocument": "default",
                     "allChangesForCluster": true
                   }
                 }
@@ -523,9 +514,7 @@
               "cursor": {},
               "pipeline": [
                 {
-                  "$changeStream": {
-                    "fullDocument": "default"
-                  }
+                  "$changeStream": {}
                 }
               ]
             },
@@ -611,9 +600,7 @@
               "cursor": {},
               "pipeline": [
                 {
-                  "$changeStream": {
-                    "fullDocument": "default"
-                  }
+                  "$changeStream": {}
                 }
               ]
             },
@@ -665,9 +652,7 @@
               "cursor": {},
               "pipeline": [
                 {
-                  "$changeStream": {
-                    "fullDocument": "default"
-                  }
+                  "$changeStream": {}
                 }
               ]
             },
@@ -756,9 +741,7 @@
               },
               "pipeline": [
                 {
-                  "$changeStream": {
-                    "fullDocument": "default"
-                  }
+                  "$changeStream": {}
                 }
               ]
             },
diff --git a/data/change-streams/change-streams.yml b/data/change-streams/change-streams.yml
diff --git a/mongo/change_stream.go b/mongo/change_stream.go

Original file line number	Diff line number	Diff line change
`@@ -33,9 +33,7 @@`
`33`	`33`	`"cursor": {},`
`34`	`34`	`"pipeline": [`
`35`	`35`	`{`
`36`		`- "$changeStream": {`
`37`		`- "fullDocument": "default"`
`38`		`- }`
	`36`	`+ "$changeStream": {}`
`39`	`37`	`}`
`40`	`38`	`]`
`41`	`39`	`},`
`@@ -153,9 +151,7 @@`
`153`	`151`	`"cursor": {},`
`154`	`152`	`"pipeline": [`
`155`	`153`	`{`
`156`		`- "$changeStream": {`
`157`		`- "fullDocument": "default"`
`158`		`- }`
	`154`	`+ "$changeStream": {}`
`159`	`155`	`}`
`160`	`156`	`]`
`161`	`157`	`},`
`@@ -226,9 +222,7 @@`
`226`	`222`	`"cursor": {},`
`227`	`223`	`"pipeline": [`
`228`	`224`	`{`
`229`		`- "$changeStream": {`
`230`		`- "fullDocument": "default"`
`231`		`- }`
	`225`	`+ "$changeStream": {}`
`232`	`226`	`},`
`233`	`227`	`{`
`234`	`228`	`"$match": {`
`@@ -312,9 +306,7 @@`
`312`	`306`	`"cursor": {},`
`313`	`307`	`"pipeline": [`
`314`	`308`	`{`
`315`		`- "$changeStream": {`
`316`		`- "fullDocument": "default"`
`317`		`- }`
	`309`	`+ "$changeStream": {}`
`318`	`310`	`}`
`319`	`311`	`]`
`320`	`312`	`},`
`@@ -404,7 +396,6 @@`
`404`	`396`	`"pipeline": [`
`405`	`397`	`{`
`406`	`398`	`"$changeStream": {`
`407`		`- "fullDocument": "default",`
`408`	`399`	`"allChangesForCluster": true`
`409`	`400`	`}`
`410`	`401`	`}`
`@@ -523,9 +514,7 @@`
`523`	`514`	`"cursor": {},`
`524`	`515`	`"pipeline": [`
`525`	`516`	`{`
`526`		`- "$changeStream": {`
`527`		`- "fullDocument": "default"`
`528`		`- }`
	`517`	`+ "$changeStream": {}`
`529`	`518`	`}`
`530`	`519`	`]`
`531`	`520`	`},`
`@@ -611,9 +600,7 @@`
`611`	`600`	`"cursor": {},`
`612`	`601`	`"pipeline": [`
`613`	`602`	`{`
`614`		`- "$changeStream": {`
`615`		`- "fullDocument": "default"`
`616`		`- }`
	`603`	`+ "$changeStream": {}`
`617`	`604`	`}`
`618`	`605`	`]`
`619`	`606`	`},`
`@@ -665,9 +652,7 @@`
`665`	`652`	`"cursor": {},`
`666`	`653`	`"pipeline": [`
`667`	`654`	`{`
`668`		`- "$changeStream": {`
`669`		`- "fullDocument": "default"`
`670`		`- }`
	`655`	`+ "$changeStream": {}`
`671`	`656`	`}`
`672`	`657`	`]`
`673`	`658`	`},`
`@@ -756,9 +741,7 @@`
`756`	`741`	`},`
`757`	`742`	`"pipeline": [`
`758`	`743`	`{`
`759`		`- "$changeStream": {`
`760`		`- "fullDocument": "default"`
`761`		`- }`
	`744`	`+ "$changeStream": {}`
`762`	`745`	`}`
`763`	`746`	`]`
`764`	`747`	`},`